feat: Regellauf mit Lock, Lebenszeichen und Zeitlimit

RuleRunner prüft Anbindungen vor jedem Lauf, baut das Ziel auf und führt
die Engine im Worker mit Inaktivitäts-Watchdog aus. Der Lock wird erst
beim echten Ende des Workers freigegeben.

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

CHANGELOG.md

@@ -34,6 +34,26 @@ Die Version steht an genau einer Stelle: `src/netmover/__init__.py`.
   wachsendem Abstand neu eingehängt.
 - Mountpoints ausschließlich unter `/mnt/netmover/<slug>`, müssen vor dem
   Einhängen leer sein und werden nach dem Aushängen entfernt.
+- Mover-Engine: Stabilitätsprüfung (Größe **und** mtime nach dem
+  Stabilitätsfenster), unfertige/versteckte Dateien und Synology-Ordner
+  (`@eaDir`, `#recycle`) werden ignoriert, Glob (mehrere Muster mit `;`) oder Regex.
+- Atomares Schreiben: `.netmover-<uuid>.part` → `fsync` → Größenprüfung →
+  Metadaten → Quellprüfung → Umbenennen ohne Überschreiben (`link`, Rückfall auf
+  `rename`) → erst dann Quelle löschen; gleiches Gerät ohne Kopie per `link`.
+- Namenskollisionen: umbenennen (`scan-1.pdf`, `scan-2.pdf`, …, jedes Mal gegen
+  das Dateisystem geprüft), überspringen oder überschreiben.
+- Buchführung im Kopiermodus (`copied_files`); verhindert auch Duplikate, wenn im
+  Verschiebemodus die Quelle nicht gelöscht werden kann.
+- Platzprüfung vor jeder Kopie (Größe + 5 %), Übernahme von mtime und Modus,
+  optional Eigentümer; Metadaten-Fehler sind Hinweise, keine Fehler; `chown` auf
+  CIFS und auf NFS mit Root-Squash wird übersprungen und einmalig gemeldet.
+- Quarantäneordner für Dateien, die wiederholt scheitern.
+- SFTP-Ziel mit demselben atomaren Ablauf (`rename` ohne Überschreiben,
+  `posix-rename` beim Überschreiben, `fsync`/`statvfs` über OpenSSH-Erweiterungen).
+- Regellauf mit Lock pro Regel, Lebenszeichen der Anbindungen und Worker mit
+  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).
 
 ### Behoben
 - Löschen von Zugangsdaten, die noch von einem Mount benutzt werden, trennt den

DEVLOG.md

@@ -9,10 +9,52 @@ 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       |
-| **Summe**  |       |       | **0:19** |      |                                        |
+| 2026-10-10 | 18:53 | 19:07 | 0:14  | 3       | Mover-Logik                            |
+| **Summe**  |       |       | **0:33** |      |                                        |
 
 ---
 
+## 2026-10-10 · 18:53–19:07 · Schritt 3: Mover-Logik
+
+**Erledigt**
+- Remote `origin` (Gitea) war eingerichtet → `main` gepusht.
+- `mover/scan.py`, `naming.py`, `ledger.py`, `destinations.py` (lokal/Kernel-Mount
+  und SFTP), `engine.py` (Ablauf pro Lauf), `runner.py` (Lock, Lebenszeichen,
+  Watchdog, Ziel-Aufbau).
+- `timeouts.run_with_watchdog`: Inaktivitäts-Zeitlimit mit `CancelToken`.
+- 60 neue Tests (183 gesamt), u. a. Abbruch in jeder Phase, Wettlauf um
+  Dateinamen, hängender Lauf, SFTP-Ziel mit Kollision gegen echten SFTP-Server.
+- Mutationsprobe: 9 eingebaute Fehler (keine Stabilitätsprüfung, kein fsync,
+  keine Quellprüfung, Wettlauf, direkt unter Endnamen schreiben, keine
+  Buchführung, keine Platzprüfung, Lock-Freigabe, kein Watchdog). Zwei überlebten
+  zunächst (fsync, enges Wettlauf-Fenster) → zwei Tests ergänzt; jetzt alle erkannt.
+
+**Entscheidungen**
+- Zeitlimit als **Inaktivitäts**-Watchdog statt Gesamtlaufzeit: große Dateien
+  über langsame Leitungen werden nicht abgebrochen, hängende Zugriffe schon.
+- Regel-Lock gibt erst der Worker selbst frei. Nach einer Zeitüberschreitung
+  bleibt die Regel blockiert, bis der hängende Worker endet – so kopieren nie
+  zwei Läufe dieselbe Datei gleichzeitig. Statusseite zeigt „läuft seit …“.
+- Stabilität innerhalb eines Laufs (zweite Messung nach dem Fenster) plus
+  Quellprüfung direkt vor dem Umbenennen.
+- Umbenennen ohne Überschreiben per `link()` (atomar, EEXIST bei Kollision);
+  ohne Hardlink-Unterstützung (oft CIFS) Rückfall auf „prüfen + rename“.
+- Bei Rekursion wird die Ordnerstruktur im Ziel beibehalten; Trennung nach
+  Quell-Unterordnern geht über mehrere Regeln mit eigenem Unterpfad.
+- Quarantänepfad ist relativ zur Quell-Anbindung (gleiches Dateisystem → rename).
+- `copied_files` wird nach jedem *vollständigen* Scan bereinigt (statt nur beim
+  Protokoll-Aufräumen) – zeitnäher, und nie nach einem abgebrochenen Scan.
+- ESTALE/EIO/ENOTCONN mitten im Lauf → Lauf abbrechen (nicht jede Datei einzeln
+  scheitern lassen, nicht gegen den Quarantäne-Zähler zählen).
+
+**Stolpersteine**
+- `tmp_path` liegt auf einem Gerät → Tests liefen erst über den Schnellweg
+  (`link`) statt über den Kopierweg; Fixture `cross_device` erzwingt ihn.
+- Prüfung „lokales Ziel nicht unter Mount-Basis“ nutzte nur `resolve()` – ein
+  Symlink umgeht das. Jetzt wörtlich **und** aufgelöst.
+- Lock-Übergabe an den Worker: doppelte Freigabe bei sehr schnellem Lauf
+  (beim Gegenlesen gefunden, vor dem ersten Testlauf behoben).
+
 ## 2026-10-10 · 18:35–18:48 · Schritt 2: Mount-Schicht, Validierung, SFTP
 
 **Anforderungsänderung (Rücksprache)**

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 2 von 6). Vollständige Doku folgt mit Schritt 6.
+> Status: in Entwicklung (Schritt 3 von 6). Vollständige Doku folgt mit Schritt 6.
 
 ## Entwicklung
 

src/netmover/mounting/manager.py

@@ -585,3 +585,10 @@ class MountManager:
         except (OSError, OperationTimeout):
             return None
         return st.f_blocks * st.f_frsize, st.f_bavail * st.f_frsize
+
+    def sftp_config_for(self, mount_id: int) -> SftpConfig:
+        """Entschlüsselte Verbindungsdaten einer SFTP-Anbindung für einen Regellauf."""
+        m, cred = self._load(mount_id)
+        if m.type is not MountType.SFTP:
+            raise SftpError(f"„{m.name}“ ist keine SFTP-Anbindung")
+        return sftp_config(MountConfig.from_model(m), cred)

src/netmover/mover/__init__.py

@@ -0,0 +1 @@
+"""Mover: Dateien finden, auf Stabilität prüfen und sicher ans Ziel bringen."""

src/netmover/mover/destinations.py

@@ -0,0 +1,489 @@
+"""Ziele: lokales Verzeichnis bzw. Kernel-Mount, oder SFTP.
+
+Fallstrick 3 – nie eine wachsende Datei unter ihrem Endnamen, nie beide Kopien
+verlieren. Ablauf für jede Datei:
+
+1. in ``<ziel>/.netmover-<uuid>.part`` schreiben, ``flush`` + ``fsync``,
+   Größe prüfen,
+2. Metadaten setzen (mtime, Modus, Eigentümer – Fehler sind nur Hinweise),
+3. prüfen, dass sich die Quelle seit der Stabilitätsmessung nicht verändert hat,
+4. atomar auf den Endnamen umbenennen – ohne zu überschreiben (``link`` bzw.
+   SFTP-``rename``), außer „überschreiben“ ist ausdrücklich eingestellt,
+5. erst danach löscht der Aufrufer die Quelle.
+
+Scheitert irgendein Schritt vor 4, wird nur die ``.part``-Datei entfernt; die
+Quelle ist unangetastet. Bricht der Prozess hart ab, bleibt höchstens eine
+``.part``-Datei zurück, die :meth:`Destination.prepare` später aufräumt.
+"""
+
+from __future__ import annotations
+
+import contextlib
+import errno
+import os
+import posixpath
+import stat
+import time
+import uuid
+from abc import ABC, abstractmethod
+from collections.abc import Callable
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Literal
+
+from netmover.models import CollisionPolicy
+from netmover.mounting.sftp import SftpSession
+from netmover.mover.naming import collision_candidates
+from netmover.timeouts import CancelToken
+
+CHUNK = 1024 * 1024
+STALE_PART_SECONDS = 3600
+PART_PREFIX = ".netmover-"
+PART_SUFFIX = ".part"
+
+FsKind = Literal["local", "cifs", "nfs", "sftp"]
+
+# Fehler, bei denen ``link`` nicht geht und wir auf „prüfen + rename“ ausweichen.
+_NO_LINK_ERRNOS = frozenset(
+    {errno.EPERM, errno.ENOTSUP, errno.EOPNOTSUPP, errno.EMLINK, errno.EXDEV, errno.ENOSYS}
+)
+
+
+class TransferError(Exception):
+    """Übertragung dieser Datei gescheitert (Meldung im Klartext)."""
+
+
+class SourceChanged(TransferError):
+    """Quelle hat sich seit der Stabilitätsmessung verändert – später erneut versuchen."""
+
+
+class CollisionSkip(Exception):
+    """Ziel existiert und die Regel sagt „überspringen“."""
+
+
+@dataclass(frozen=True)
+class MetaOptions:
+    atime_ns: int
+    mtime_ns: int
+    mode: int | None  # Rechte-Bits, None = nicht setzen
+    owner: tuple[int, int] | None  # None = nicht setzen
+
+
+@dataclass
+class CommitResult:
+    final_rel: str
+    notes: list[str] = field(default_factory=list)
+    source_consumed: bool = False  # True: Quelle wurde per rename bereits „verschoben“
+
+
+Verify = Callable[[], None]
+
+
+class Destination(ABC):
+    kind: FsKind
+
+    @abstractmethod
+    def describe(self, rel: str) -> str:
+        """Lesbarer Zielpfad für Protokoll und Probelauf."""
+
+    @abstractmethod
+    def prepare(self) -> None:
+        """Vor dem Lauf: Zielordner prüfen, alte ``.part``-Reste entfernen."""
+
+    @abstractmethod
+    def exists(self, rel: str) -> bool: ...
+
+    @abstractmethod
+    def free_bytes(self) -> int | None: ...
+
+    def close(self) -> None:  # noqa: B027 – absichtlich optional
+        pass
+
+    # -- Primitive der Unterklassen -------------------------------------------------
+
+    @abstractmethod
+    def _ensure_parent(self, rel: str) -> None: ...
+
+    @abstractmethod
+    def _upload(self, src: Path, tmp: str, size: int, token: CancelToken) -> list[str]: ...
+
+    @abstractmethod
+    def _apply_meta(self, rel: str, meta: MetaOptions) -> list[str]: ...
+
+    @abstractmethod
+    def _link_noreplace(self, tmp: str, final: str) -> None:
+        """Atomar umbenennen; ``FileExistsError``, wenn ``final`` existiert."""
+
+    @abstractmethod
+    def _replace(self, tmp: str, final: str) -> list[str]: ...
+
+    @abstractmethod
+    def _remove(self, rel: str) -> None: ...
+
+    def _sync_dir(self, rel_dir: str) -> None:  # noqa: B027
+        pass
+
+    # -- gemeinsamer Ablauf ---------------------------------------------------------
+
+    def plan(self, rel: str, policy: CollisionPolicy) -> str | None:
+        """Voraussichtlicher Endname (Probelauf / Vorabprüfung); None = überspringen."""
+        if policy is CollisionPolicy.OVERWRITE:
+            return rel
+        if policy is CollisionPolicy.SKIP:
+            return None if self.exists(rel) else rel
+        for cand in collision_candidates(rel):
+            if not self.exists(cand):
+                return cand
+        raise TransferError(f"Kein freier Dateiname für {rel}")
+
+    def commit(
+        self,
+        src: Path,
+        size: int,
+        rel: str,
+        policy: CollisionPolicy,
+        meta: MetaOptions,
+        verify: Verify,
+        token: CancelToken,
+    ) -> CommitResult:
+        self._ensure_parent(rel)
+        tmp = posixpath.join(
+            posixpath.dirname(rel), f"{PART_PREFIX}{uuid.uuid4().hex}{PART_SUFFIX}"
+        )
+        try:
+            notes = self._upload(src, tmp, size, token)
+            notes += self._apply_meta(tmp, meta)
+            verify()
+            token.check()
+            final, more = self._finalize(tmp, rel, policy)
+            notes += more
+        except BaseException:
+            with contextlib.suppress(Exception):
+                self._remove(tmp)
+            raise
+        self._sync_dir(posixpath.dirname(final))
+        return CommitResult(final, notes)
+
+    def move_same_device(
+        self, src: Path, rel: str, policy: CollisionPolicy, meta: MetaOptions
+    ) -> CommitResult | None:
+        """Schneller Weg ohne Kopie, wenn Quelle und Ziel auf demselben Gerät liegen."""
+        return None
+
+    def _finalize(self, tmp: str, rel: str, policy: CollisionPolicy) -> tuple[str, list[str]]:
+        if policy is CollisionPolicy.OVERWRITE:
+            return rel, self._replace(tmp, rel)
+        candidates = collision_candidates(rel) if policy is CollisionPolicy.RENAME else iter([rel])
+        for cand in candidates:
+            if self.exists(cand):
+                if policy is CollisionPolicy.SKIP:
+                    raise CollisionSkip
+                continue
+            try:
+                self._link_noreplace(tmp, cand)
+            except FileExistsError:
+                # Wettlauf: jemand anderes hat den Namen gerade belegt
+                if policy is CollisionPolicy.SKIP:
+                    raise CollisionSkip from None
+                continue
+            return cand, []
+        raise TransferError(f"Kein freier Dateiname für {rel}")
+
+
+class LocalDestination(Destination):
+    """Lokales Verzeichnis oder Verzeichnis auf einem (geprüften) Kernel-Mount."""
+
+    def __init__(
+        self,
+        root: Path,
+        kind: Literal["local", "cifs", "nfs"] = "local",
+        *,
+        create_root: bool = False,
+    ) -> None:
+        self.root = root
+        self.kind = kind
+        self.create_root = create_root
+
+    def _p(self, rel: str) -> Path:
+        return self.root / rel
+
+    def describe(self, rel: str) -> str:
+        return str(self._p(rel))
+
+    def prepare(self) -> None:
+        if not self.root.is_dir():
+            if not self.create_root:
+                raise TransferError(f"Zielordner {self.root} existiert nicht")
+            self.root.mkdir(mode=0o755, parents=True, exist_ok=True)
+        cutoff = time.time() - STALE_PART_SECONDS
+        for entry in os.scandir(self.root):
+            if entry.name.startswith(PART_PREFIX) and entry.name.endswith(PART_SUFFIX):
+                with contextlib.suppress(OSError):
+                    if entry.stat(follow_symlinks=False).st_mtime < cutoff:
+                        os.unlink(entry.path)
+
+    def exists(self, rel: str) -> bool:
+        return os.path.lexists(self._p(rel))
+
+    def free_bytes(self) -> int | None:
+        st = os.statvfs(self.root)
+        return st.f_bavail * st.f_frsize
+
+    def _ensure_parent(self, rel: str) -> None:
+        self._p(rel).parent.mkdir(mode=0o755, parents=True, exist_ok=True)
+
+    def _upload(self, src: Path, tmp: str, size: int, token: CancelToken) -> list[str]:
+        written = 0
+        fd = os.open(self._p(tmp), os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
+        try:
+            with src.open("rb") as fin:
+                while chunk := fin.read(CHUNK):
+                    token.check()
+                    view = memoryview(chunk)
+                    while view:
+                        n = os.write(fd, view)
+                        view = view[n:]
+                    written += len(chunk)
+            if written != size:
+                raise SourceChanged(
+                    "Quelle hat sich während der Übertragung geändert "
+                    f"({written} statt {size} Bytes)"
+                )
+            os.fsync(fd)
+            if os.fstat(fd).st_size != size:
+                raise TransferError("Größe der Zieldatei stimmt nach dem Schreiben nicht")
+        finally:
+            os.close(fd)
+        return []
+
+    def _apply_meta(self, rel: str, meta: MetaOptions) -> list[str]:
+        notes: list[str] = []
+        path = self._p(rel)
+        if self.kind != "cifs":  # auf CIFS bestimmen die Mount-Optionen Eigentümer/Rechte
+            if meta.owner is not None:
+                try:
+                    os.chown(path, *meta.owner)
+                except PermissionError:
+                    notes.append(
+                        f"Eigentümer {meta.owner[0]}:{meta.owner[1]} nicht gesetzt – keine "
+                        "Berechtigung (läuft NetMover als root? Root-Squash?)"
+                    )
+                except OSError as exc:
+                    notes.append(f"Eigentümer nicht gesetzt: {exc.strerror}")
+            if meta.mode is not None:
+                try:
+                    os.chmod(path, meta.mode)
+                except OSError as exc:
+                    notes.append(f"Modus {meta.mode:04o} nicht gesetzt: {exc.strerror}")
+        try:
+            os.utime(path, ns=(meta.atime_ns, meta.mtime_ns))
+        except OSError as exc:
+            notes.append(f"Änderungszeit nicht übernommen: {exc.strerror}")
+        return notes
+
+    def _link_noreplace(self, tmp: str, final: str) -> None:
+        src, dst = self._p(tmp), self._p(final)
+        try:
+            os.link(src, dst)
+        except FileExistsError:
+            raise
+        except OSError as exc:
+            if exc.errno not in _NO_LINK_ERRNOS:
+                raise
+            # Dateisystem ohne Hardlinks (häufig CIFS): prüfen + rename.
+            # Kleines Wettlauf-Fenster, aber nie eine halbe Datei unter dem Endnamen.
+            if os.path.lexists(dst):
+                raise FileExistsError(errno.EEXIST, "exists", str(dst)) from None
+            os.rename(src, dst)
+            return
+        with contextlib.suppress(OSError):
+            os.unlink(src)  # Rest wird sonst später als alte .part-Datei aufgeräumt
+
+    def _replace(self, tmp: str, final: str) -> list[str]:
+        os.replace(self._p(tmp), self._p(final))
+        return []
+
+    def _remove(self, rel: str) -> None:
+        with contextlib.suppress(FileNotFoundError):
+            os.unlink(self._p(rel))
+
+    def _sync_dir(self, rel_dir: str) -> None:
+        with contextlib.suppress(OSError):
+            fd = os.open(self._p(rel_dir) if rel_dir else self.root, os.O_RDONLY | os.O_DIRECTORY)
+            try:
+                os.fsync(fd)
+            finally:
+                os.close(fd)
+
+    def move_same_device(
+        self, src: Path, rel: str, policy: CollisionPolicy, meta: MetaOptions
+    ) -> CommitResult | None:
+        try:
+            same = os.stat(src).st_dev == os.stat(self.root).st_dev
+        except OSError:
+            return None
+        if not same:
+            return None
+        self._ensure_parent(rel)
+        consumed = True
+        if policy is CollisionPolicy.OVERWRITE:
+            os.replace(src, self._p(rel))
+            final = rel
+        else:
+            final = ""
+            candidates = (
+                collision_candidates(rel) if policy is CollisionPolicy.RENAME else iter([rel])
+            )
+            for cand in candidates:
+                if self.exists(cand):
+                    if policy is CollisionPolicy.SKIP:
+                        raise CollisionSkip
+                    continue
+                try:
+                    os.link(src, self._p(cand))
+                except FileExistsError:
+                    if policy is CollisionPolicy.SKIP:
+                        raise CollisionSkip from None
+                    continue
+                except OSError as exc:
+                    if exc.errno not in _NO_LINK_ERRNOS:
+                        raise
+                    if self.exists(cand):
+                        continue
+                    os.rename(src, self._p(cand))
+                    final = cand
+                    break
+                # Ziel ist vollständig da; den alten Namen entfernt der Aufrufer –
+                # mit derselben Fehlerbehandlung wie nach einer Kopie.
+                consumed = False
+                final = cand
+                break
+            if not final:
+                raise TransferError(f"Kein freier Dateiname für {rel}")
+        notes = self._apply_meta(final, meta)
+        self._sync_dir(posixpath.dirname(final))
+        return CommitResult(final, notes, source_consumed=consumed)
+
+
+class SftpDestination(Destination):
+    kind: FsKind = "sftp"
+
+    def __init__(self, session: SftpSession, base: str, label: str = "") -> None:
+        self.session = session
+        self.client = session.client
+        self.base = base
+        self.label = label
+
+    def _p(self, rel: str) -> str:
+        return posixpath.join(self.base, rel) if rel else self.base
+
+    def describe(self, rel: str) -> str:
+        prefix = f"{self.label}:" if self.label else "sftp:"
+        return prefix + self._p(rel)
+
+    def close(self) -> None:
+        self.session.close()
+
+    def prepare(self) -> None:
+        try:
+            attrs = self.client.stat(self.base)
+        except OSError as exc:
+            raise TransferError(f"SFTP-Zielordner {self.base}: {exc}") from None
+        if attrs.st_mode is None or not stat.S_ISDIR(attrs.st_mode):
+            raise TransferError(f"SFTP-Ziel {self.base} ist kein Verzeichnis")
+        cutoff = time.time() - STALE_PART_SECONDS
+        for a in self.client.listdir_attr(self.base):
+            is_part = a.filename.startswith(PART_PREFIX) and a.filename.endswith(PART_SUFFIX)
+            if is_part and a.st_mtime is not None and a.st_mtime < cutoff:
+                with contextlib.suppress(OSError):
+                    self.client.remove(self._p(a.filename))
+
+    def exists(self, rel: str) -> bool:
+        try:
+            self.client.lstat(self._p(rel))
+        except FileNotFoundError:
+            return False
+        return True
+
+    def free_bytes(self) -> int | None:
+        usage = self.session.statvfs(self.base)
+        return usage[1] if usage else None
+
+    def _ensure_parent(self, rel: str) -> None:
+        parent = posixpath.dirname(rel)
+        if not parent:
+            return
+        current = self.base
+        for part in parent.split("/"):
+            current = posixpath.join(current, part)
+            try:
+                self.client.stat(current)
+            except FileNotFoundError:
+                self.client.mkdir(current, mode=0o755)
+
+    def _upload(self, src: Path, tmp: str, size: int, token: CancelToken) -> list[str]:
+        notes: list[str] = []
+        written = 0
+        with self.client.open(self._p(tmp), "wb") as fh, src.open("rb") as fin:
+            fh.set_pipelined(True)
+            while chunk := fin.read(CHUNK // 4):
+                token.check()
+                fh.write(chunk)
+                written += len(chunk)
+            fh.flush()
+            if written != size:
+                raise SourceChanged(
+                    "Quelle hat sich während der Übertragung geändert "
+                    f"({written} statt {size} Bytes)"
+                )
+            if not self.session.fsync(fh):
+                notes.append("SFTP-Server unterstützt kein fsync")
+        remote = self.client.stat(self._p(tmp)).st_size
+        if remote != size:
+            raise TransferError(f"Zieldatei hat {remote} statt {size} Bytes")
+        return notes
+
+    def _apply_meta(self, rel: str, meta: MetaOptions) -> list[str]:
+        notes: list[str] = []
+        path = self._p(rel)
+        if meta.owner is not None:
+            try:
+                self.client.chown(path, *meta.owner)
+            except OSError as exc:
+                notes.append(f"Eigentümer per SFTP nicht gesetzt: {exc}")
+        if meta.mode is not None:
+            try:
+                self.client.chmod(path, meta.mode)
+            except OSError as exc:
+                notes.append(f"Modus per SFTP nicht gesetzt: {exc}")
+        try:
+            self.client.utime(path, (meta.atime_ns / 1e9, meta.mtime_ns / 1e9))
+        except OSError as exc:
+            notes.append(f"Änderungszeit nicht übernommen: {exc}")
+        return notes
+
+    def _link_noreplace(self, tmp: str, final: str) -> None:
+        if self.exists(final):
+            raise FileExistsError(errno.EEXIST, "exists", final)
+        try:
+            # SFTPv3-rename überschreibt nicht (OpenSSH: link + unlink).
+            self.client.rename(self._p(tmp), self._p(final))
+        except OSError:
+            if self.exists(final):
+                raise FileExistsError(errno.EEXIST, "exists", final) from None
+            raise
+
+    def _replace(self, tmp: str, final: str) -> list[str]:
+        try:
+            self.client.posix_rename(self._p(tmp), self._p(final))
+        except OSError:
+            # Server ohne posix-rename: löschen + umbenennen (kurz kein Ziel, nie halb)
+            with contextlib.suppress(FileNotFoundError):
+                self.client.remove(self._p(final))
+            self.client.rename(self._p(tmp), self._p(final))
+            return ["Server kann nicht atomar überschreiben – Ziel kurz nicht vorhanden"]
+        return []
+
+    def _remove(self, rel: str) -> None:
+        with contextlib.suppress(FileNotFoundError):
+            self.client.remove(self._p(rel))

src/netmover/mover/engine.py

@@ -0,0 +1,434 @@
+"""Ablauf eines Regellaufs – ohne Scheduler, ohne Datenbank-Protokoll.
+
+Reihenfolge pro Lauf:
+
+1. Ziel vorbereiten, Quelle scannen (Filter, versteckte/unfertige Dateien raus),
+2. Kopier-Buchführung: bereits übertragene Dateiversionen still auslassen,
+3. Größe und Mindestalter prüfen,
+4. Stabilität: nach dem Stabilitätsfenster erneut messen – nur Dateien mit
+   unveränderter Größe **und** mtime gehen weiter (Fallstrick 1),
+5. pro Datei: ggf. Quarantäne, Kollisionsregel, Platzprüfung (Größe + 5 %,
+   Fallstrick 14), atomare Übertragung, erst dann Quelle löschen.
+"""
+
+from __future__ import annotations
+
+import errno
+import os
+import stat
+import time
+from collections.abc import Callable
+from dataclasses import dataclass, field
+from pathlib import Path
+
+from netmover.logging_setup import event, get_logger
+from netmover.models import (
+    CollisionPolicy,
+    ItemStatus,
+    PatternType,
+    RuleAction,
+    RunStatus,
+)
+from netmover.mover.destinations import (
+    CollisionSkip,
+    Destination,
+    LocalDestination,
+    MetaOptions,
+    SourceChanged,
+    TransferError,
+)
+from netmover.mover.ledger import FailureInfo, Ledger
+from netmover.mover.naming import collision_candidates
+from netmover.mover.scan import Candidate, FileState, Matcher, current_state, scan
+from netmover.timeouts import Cancelled, CancelToken
+
+log = get_logger(__name__)
+
+SPACE_RESERVE = 1.05
+# Fehler, die auf eine gestörte Anbindung hindeuten: Lauf abbrechen statt jede
+# weitere Datei scheitern zu lassen – der nächste Lauf prüft und hängt neu ein.
+MOUNT_ERRNOS = frozenset({errno.ESTALE, errno.EIO, errno.ENOTCONN, errno.EHOSTDOWN})
+
+
+class RunAborted(Exception):
+    pass
+
+
+@dataclass(frozen=True)
+class RuleSnapshot:
+    id: int
+    name: str
+    action: RuleAction = RuleAction.MOVE
+    on_collision: CollisionPolicy = CollisionPolicy.RENAME
+    pattern: str = "*"
+    pattern_type: PatternType = PatternType.GLOB
+    recursive: bool = False
+    stability_seconds: int = 5
+    min_age_seconds: int = 0
+    max_size_mb: int | None = None
+    set_owner: tuple[int, int] | None = None
+    set_mode: int | None = None
+    quarantine_after: int = 3
+    dst_root_squash: bool | None = None  # nur NFS-Ziele: Ergebnis des Schreibtests
+
+
+@dataclass
+class ItemResult:
+    src: str
+    status: ItemStatus
+    dst: str | None = None
+    size: int | None = None
+    message: str | None = None
+
+
+@dataclass
+class RunResult:
+    status: RunStatus = RunStatus.OK
+    dry_run: bool = False
+    seen: int = 0
+    moved: int = 0
+    skipped: int = 0
+    failed: int = 0
+    message: str | None = None
+    items: list[ItemResult] = field(default_factory=list)
+
+    def add(self, item: ItemResult) -> None:
+        self.items.append(item)
+        if item.status in (ItemStatus.MOVED, ItemStatus.COPIED, ItemStatus.PLANNED):
+            self.moved += 1
+        elif item.status in (ItemStatus.SKIPPED, ItemStatus.QUARANTINED):
+            self.skipped += 1
+        else:
+            self.failed += 1
+
+
+def _describe_os_error(exc: OSError) -> str:
+    where = f" ({exc.filename})" if exc.filename else ""
+    return f"{exc.strerror or exc}{where}"
+
+
+def _mb(n: int) -> str:
+    return f"{n / 1024 / 1024:.1f} MB"
+
+
+class Engine:
+    def __init__(
+        self,
+        ledger: Ledger,
+        *,
+        sleep: Callable[[float], None] = time.sleep,
+        clock: Callable[[], float] = time.time,
+        warn_once: Callable[[str], bool] | None = None,
+    ) -> None:
+        self.ledger = ledger
+        self._sleep = sleep
+        self._clock = clock
+        self._warned: set[str] = set()
+        self._warn_once = warn_once or self._default_warn_once
+
+    def _default_warn_once(self, key: str) -> bool:
+        if key in self._warned:
+            return False
+        self._warned.add(key)
+        return True
+
+    # -- Lauf -------------------------------------------------------------------------
+
+    def execute(
+        self,
+        rule: RuleSnapshot,
+        src_root: Path,
+        dest: Destination,
+        token: CancelToken,
+        *,
+        dry_run: bool = False,
+        quarantine_root: Path | None = None,
+    ) -> RunResult:
+        result = RunResult(dry_run=dry_run)
+        try:
+            self._execute(rule, src_root, dest, token, result, dry_run, quarantine_root)
+        except Cancelled:
+            result.status = RunStatus.ERROR
+            result.message = "Lauf abgebrochen"
+        except RunAborted as exc:
+            result.status = RunStatus.ERROR
+            result.message = str(exc)
+        except TransferError as exc:
+            result.status = RunStatus.ERROR
+            result.message = str(exc)
+        except OSError as exc:
+            result.status = RunStatus.ERROR
+            result.message = _describe_os_error(exc)
+        else:
+            if result.failed:
+                result.status = RunStatus.PARTIAL
+        return result
+
+    def _execute(
+        self,
+        rule: RuleSnapshot,
+        src_root: Path,
+        dest: Destination,
+        token: CancelToken,
+        result: RunResult,
+        dry_run: bool,
+        quarantine_root: Path | None,
+    ) -> None:
+        if not src_root.is_dir():
+            raise RunAborted(f"Quellordner {src_root} existiert nicht")
+        if not dry_run:
+            dest.prepare()
+        token.check()
+
+        exclude: set[Path] = set()
+        if quarantine_root is not None:
+            exclude.add(quarantine_root.resolve())
+        if isinstance(dest, LocalDestination):
+            exclude.add(dest.root.resolve())  # Ziel innerhalb der Quelle nie erneut scannen
+        matcher = Matcher(rule.pattern, rule.pattern_type)
+        candidates = list(scan(src_root, matcher, rule.recursive, frozenset(exclude)))
+        result.seen = len(candidates)
+        token.check()
+
+        # Kopier-Buchführung (Fallstrick 11). Gilt auch für „verschieben“, falls
+        # dort einmal die Quelle nicht gelöscht werden konnte.
+        done = self.ledger.copied_keys(rule.id)
+        if not dry_run:
+            self.ledger.prune_copied(rule.id, {c.relpath for c in candidates})
+        fresh = [c for c in candidates if (c.relpath, c.state.size, c.state.mtime_ns) not in done]
+        unchanged = len(candidates) - len(fresh)
+
+        pending: list[Candidate] = []
+        now = self._clock()
+        for c in fresh:
+            if rule.max_size_mb is not None and c.state.size > rule.max_size_mb * 1024 * 1024:
+                result.add(
+                    ItemResult(
+                        c.relpath,
+                        ItemStatus.SKIPPED,
+                        size=c.state.size,
+                        message=f"größer als {rule.max_size_mb} MB",
+                    )
+                )
+            elif rule.min_age_seconds and now - c.state.mtime_ns / 1e9 < rule.min_age_seconds:
+                result.add(
+                    ItemResult(
+                        c.relpath,
+                        ItemStatus.SKIPPED,
+                        size=c.state.size,
+                        message=f"jünger als {rule.min_age_seconds} s",
+                    )
+                )
+            else:
+                pending.append(c)
+
+        # Stabilität (Fallstrick 1): zweite Messung nach dem Stabilitätsfenster.
+        if pending and rule.stability_seconds > 0:
+            self._sleep(rule.stability_seconds)
+            token.check()
+        stable: list[Candidate] = []
+        for c in pending:
+            st = current_state(c.path)
+            if st is None:
+                continue  # inzwischen verschwunden – nichts zu tun
+            if st != c.state:
+                result.add(
+                    ItemResult(
+                        c.relpath,
+                        ItemStatus.SKIPPED,
+                        size=st.size,
+                        message="wird noch geschrieben (Größe/Zeit geändert)",
+                    )
+                )
+                continue
+            stable.append(c)
+
+        failures = self.ledger.failures(rule.id)
+        for c in stable:
+            token.check()
+            result.add(self._process(rule, c, dest, token, dry_run, quarantine_root, failures))
+
+        if unchanged:
+            result.message = f"{unchanged} bereits übertragen"
+
+    # -- einzelne Datei ------------------------------------------------------------------
+
+    def _meta(
+        self, rule: RuleSnapshot, path: Path, dest: Destination
+    ) -> tuple[MetaOptions, list[str]]:
+        st = os.stat(path)
+        notes: list[str] = []
+        owner = rule.set_owner
+        if owner is not None:
+            if dest.kind == "cifs":
+                owner = None
+                if self._warn_once(f"{rule.id}:cifs-chown"):
+                    notes.append(
+                        "CIFS-Ziel: Eigentümer bestimmen die Mount-Optionen, chown übersprungen"
+                    )
+            elif dest.kind == "nfs" and rule.dst_root_squash is not False:
+                owner = None
+                if self._warn_once(f"{rule.id}:nfs-chown"):
+                    reason = "Root-Squash aktiv" if rule.dst_root_squash else "Schreibtest fehlt"
+                    notes.append(f"NFS-Ziel: chown übersprungen ({reason})")
+        mode = rule.set_mode if rule.set_mode is not None else stat.S_IMODE(st.st_mode)
+        return MetaOptions(st.st_atime_ns, st.st_mtime_ns, mode, owner), notes
+
+    def _process(
+        self,
+        rule: RuleSnapshot,
+        c: Candidate,
+        dest: Destination,
+        token: CancelToken,
+        dry_run: bool,
+        quarantine_root: Path | None,
+        failures: dict[str, FailureInfo],
+    ) -> ItemResult:
+        size = c.state.size
+        failure = failures.get(c.relpath)
+
+        # Quarantäne für wiederholt scheiternde Dateien
+        if (
+            quarantine_root is not None
+            and failure is not None
+            and failure.attempts >= rule.quarantine_after
+            and (failure.size, failure.mtime_ns) == (c.state.size, c.state.mtime_ns)
+        ):
+            if dry_run:
+                return ItemResult(
+                    c.relpath,
+                    ItemStatus.PLANNED,
+                    str(quarantine_root / c.relpath),
+                    size,
+                    "würde in Quarantäne verschoben",
+                )
+            return self._quarantine(rule, c, quarantine_root, failure.attempts)
+
+        try:
+            planned = dest.plan(c.relpath, rule.on_collision)
+        except OSError as exc:
+            return self._fail(rule, c, exc, dry_run)
+        if planned is None:
+            return ItemResult(
+                c.relpath,
+                ItemStatus.SKIPPED,
+                dest.describe(c.relpath),
+                size,
+                "existiert bereits im Ziel",
+            )
+        if dry_run:
+            msg = (
+                "würde überschreiben"
+                if (rule.on_collision is CollisionPolicy.OVERWRITE and dest.exists(planned))
+                else None
+            )
+            verb = "kopieren" if rule.action is RuleAction.COPY else "verschieben"
+            return ItemResult(
+                c.relpath, ItemStatus.PLANNED, dest.describe(planned), size, msg or f"würde {verb}"
+            )
+
+        try:
+            free = dest.free_bytes()
+        except OSError:
+            free = None
+        if free is not None and free < size * SPACE_RESERVE:
+            return ItemResult(
+                c.relpath,
+                ItemStatus.FAILED,
+                dest.describe(planned),
+                size,
+                f"zu wenig Platz am Ziel (frei {_mb(free)}, "
+                f"benötigt {_mb(int(size * SPACE_RESERVE))})",
+            )
+
+        def verify() -> None:
+            if current_state(c.path) != c.state:
+                raise SourceChanged("Quelle hat sich während der Übertragung geändert")
+
+        try:
+            meta, notes = self._meta(rule, c.path, dest)
+            res = None
+            if rule.action is RuleAction.MOVE:
+                res = dest.move_same_device(c.path, c.relpath, rule.on_collision, meta)
+            if res is None:
+                res = dest.commit(c.path, size, c.relpath, rule.on_collision, meta, verify, token)
+        except CollisionSkip:
+            return ItemResult(
+                c.relpath,
+                ItemStatus.SKIPPED,
+                dest.describe(c.relpath),
+                size,
+                "existiert bereits im Ziel",
+            )
+        except SourceChanged as exc:
+            return ItemResult(c.relpath, ItemStatus.SKIPPED, None, size, str(exc))
+        except (TransferError, OSError) as exc:
+            return self._fail(rule, c, exc, dry_run)
+
+        notes += res.notes
+        target = dest.describe(res.final_rel)
+        if c.relpath in failures:
+            self.ledger.clear_failure(rule.id, c.relpath)
+
+        if rule.action is RuleAction.COPY:
+            self.ledger.record_copy(rule.id, c.relpath, size, c.state.mtime_ns, target)
+            return ItemResult(c.relpath, ItemStatus.COPIED, target, size, "; ".join(notes) or None)
+
+        if not res.source_consumed:
+            try:
+                os.unlink(c.path)
+            except OSError as exc:
+                # Ziel ist vollständig – Buchführung verhindert Duplikate im nächsten Lauf.
+                self.ledger.record_copy(rule.id, c.relpath, size, c.state.mtime_ns, target)
+                return ItemResult(
+                    c.relpath,
+                    ItemStatus.FAILED,
+                    target,
+                    size,
+                    f"Ziel geschrieben, aber Quelle nicht löschbar: {_describe_os_error(exc)} – "
+                    "Datei wird nicht erneut übertragen",
+                )
+        return ItemResult(c.relpath, ItemStatus.MOVED, target, size, "; ".join(notes) or None)
+
+    def _fail(self, rule: RuleSnapshot, c: Candidate, exc: Exception, dry_run: bool) -> ItemResult:
+        if isinstance(exc, OSError) and exc.errno in MOUNT_ERRNOS:
+            raise RunAborted(f"Anbindung gestört: {_describe_os_error(exc)}") from exc
+        msg = _describe_os_error(exc) if isinstance(exc, OSError) else str(exc)
+        if not dry_run:
+            attempts = self.ledger.record_failure(
+                rule.id, c.relpath, c.state.size, c.state.mtime_ns, msg
+            )
+            msg = f"{msg} (Versuch {attempts})"
+        event(log, "file.failed", level=30, rule=rule.name, file=c.relpath, error=msg)
+        return ItemResult(c.relpath, ItemStatus.FAILED, None, c.state.size, msg)
+
+    def _quarantine(
+        self, rule: RuleSnapshot, c: Candidate, root: Path, attempts: int
+    ) -> ItemResult:
+        try:
+            for cand in collision_candidates(c.relpath):
+                target = root / cand
+                if os.path.lexists(target):
+                    continue
+                target.parent.mkdir(mode=0o755, parents=True, exist_ok=True)
+                os.rename(c.path, target)
+                break
+            else:
+                raise TransferError("kein freier Name im Quarantäneordner")
+        except (OSError, TransferError) as exc:
+            msg = _describe_os_error(exc) if isinstance(exc, OSError) else str(exc)
+            return ItemResult(
+                c.relpath, ItemStatus.FAILED, None, c.state.size, f"Quarantäne gescheitert: {msg}"
+            )
+        self.ledger.clear_failure(rule.id, c.relpath)
+        event(log, "file.quarantined", level=30, rule=rule.name, file=c.relpath)
+        return ItemResult(
+            c.relpath,
+            ItemStatus.QUARANTINED,
+            str(target),
+            c.state.size,
+            f"nach {attempts} Fehlversuchen in Quarantäne verschoben",
+        )
+
+
+__all__ = ["Engine", "FileState", "ItemResult", "RuleSnapshot", "RunAborted", "RunResult"]

src/netmover/mover/ledger.py

@@ -0,0 +1,108 @@
+"""Buchführung in der Datenbank.
+
+* ``copied_files`` (Fallstrick 11): Welche Dateiversion (Pfad, Größe, mtime) wurde
+  schon übertragen? Verhindert endloses Neu-Kopieren im Kopiermodus – und im
+  Verschiebemodus Duplikate, wenn die Quelle nicht gelöscht werden konnte.
+* ``file_failures``: Fehlversuche pro Datei als Grundlage der Quarantäne.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from sqlalchemy import delete, select
+from sqlalchemy.orm import Session, sessionmaker
+
+from netmover.db import session_scope
+from netmover.models import CopiedFile, FileFailure, utcnow
+
+CopyKey = tuple[str, int, int]  # (relpath, size, mtime_ns)
+
+
+@dataclass(frozen=True)
+class FailureInfo:
+    attempts: int
+    size: int
+    mtime_ns: int
+
+
+class Ledger:
+    def __init__(self, session_factory: sessionmaker[Session]) -> None:
+        self.session_factory = session_factory
+
+    def copied_keys(self, rule_id: int) -> set[CopyKey]:
+        with session_scope(self.session_factory) as s:
+            rows = s.execute(
+                select(CopiedFile.src_relpath, CopiedFile.size, CopiedFile.mtime_ns).where(
+                    CopiedFile.rule_id == rule_id
+                )
+            ).all()
+        return {(r[0], r[1], r[2]) for r in rows}
+
+    def record_copy(self, rule_id: int, relpath: str, size: int, mtime_ns: int, dst: str) -> None:
+        with session_scope(self.session_factory) as s:
+            exists = s.scalar(
+                select(CopiedFile.id).where(
+                    CopiedFile.rule_id == rule_id,
+                    CopiedFile.src_relpath == relpath,
+                    CopiedFile.size == size,
+                    CopiedFile.mtime_ns == mtime_ns,
+                )
+            )
+            if exists is None:
+                s.add(
+                    CopiedFile(
+                        rule_id=rule_id,
+                        src_relpath=relpath,
+                        size=size,
+                        mtime_ns=mtime_ns,
+                        dst_path=dst,
+                    )
+                )
+
+    def prune_copied(self, rule_id: int, present: set[str]) -> int:
+        """Entfernt Einträge, deren Quelldatei verschwunden ist. Nur nach einem
+        *vollständigen* Scan aufrufen – sonst würde alles erneut kopiert."""
+        with session_scope(self.session_factory) as s:
+            rows = s.execute(
+                select(CopiedFile.id, CopiedFile.src_relpath).where(CopiedFile.rule_id == rule_id)
+            ).all()
+            stale = [r[0] for r in rows if r[1] not in present]
+            if stale:
+                s.execute(delete(CopiedFile).where(CopiedFile.id.in_(stale)))
+            return len(stale)
+
+    def failures(self, rule_id: int) -> dict[str, FailureInfo]:
+        with session_scope(self.session_factory) as s:
+            rows = s.scalars(select(FileFailure).where(FileFailure.rule_id == rule_id)).all()
+            return {r.src_relpath: FailureInfo(r.attempts, r.size, r.mtime_ns) for r in rows}
+
+    def record_failure(
+        self, rule_id: int, relpath: str, size: int, mtime_ns: int, error: str
+    ) -> int:
+        with session_scope(self.session_factory) as s:
+            row = s.scalar(
+                select(FileFailure).where(
+                    FileFailure.rule_id == rule_id, FileFailure.src_relpath == relpath
+                )
+            )
+            if row is None:
+                row = FileFailure(
+                    rule_id=rule_id, src_relpath=relpath, size=size, mtime_ns=mtime_ns, attempts=0
+                )
+                s.add(row)
+            elif (row.size, row.mtime_ns) != (size, mtime_ns):
+                # neue Dateiversion – Zählung beginnt von vorn
+                row.size, row.mtime_ns, row.attempts = size, mtime_ns, 0
+            row.attempts += 1
+            row.last_error = error
+            row.last_attempt_at = utcnow()
+            return row.attempts
+
+    def clear_failure(self, rule_id: int, relpath: str) -> None:
+        with session_scope(self.session_factory) as s:
+            s.execute(
+                delete(FileFailure).where(
+                    FileFailure.rule_id == rule_id, FileFailure.src_relpath == relpath
+                )
+            )

src/netmover/mover/naming.py

@@ -0,0 +1,22 @@
+"""Namenskollisionen (Fallstrick 4): ``scan001.pdf`` → ``scan001-1.pdf`` → …
+
+Die Kandidaten werden nur erzeugt; ob ein Name frei ist, entscheidet jedes Mal
+das Ziel-Dateisystem (kein Vorhalten im Speicher).
+"""
+
+from __future__ import annotations
+
+import posixpath
+from collections.abc import Iterator
+
+MAX_SUFFIX = 9999
+
+
+def collision_candidates(relpath: str) -> Iterator[str]:
+    yield relpath
+    directory, name = posixpath.split(relpath)
+    stem, dot, ext = name.rpartition(".")
+    if not stem:  # kein Punkt bzw. nur führender Punkt
+        stem, dot, ext = name, "", ""
+    for n in range(1, MAX_SUFFIX + 1):
+        yield posixpath.join(directory, f"{stem}-{n}{dot}{ext}")

src/netmover/mover/runner.py

@@ -0,0 +1,272 @@
+"""Einen Regellauf sicher ausführen.
+
+* Fallstrick 8: ein Lock pro Regel; läuft sie noch, wird der Termin übersprungen.
+* Fallstrick 2: vor jedem Lauf prüft der :class:`MountManager`, dass Quelle und
+  Ziel wirklich eingehängt und lebendig sind – sonst wird nichts angefasst.
+* Fallstrick 13: die Dateizugriffe laufen in einem eigenen Worker mit
+  Inaktivitäts-Zeitlimit. Der Lock wird erst freigegeben, wenn der Worker
+  wirklich endet – ein hängender Lauf kann also nie parallel zu einem neuen
+  dieselben Dateien bearbeiten; neue Termine werden bis dahin übersprungen.
+"""
+
+from __future__ import annotations
+
+import posixpath
+import threading
+import time
+from collections.abc import Callable
+from dataclasses import dataclass
+from datetime import UTC, datetime
+from pathlib import Path
+
+from sqlalchemy.orm import Session, sessionmaker
+
+from netmover.db import session_scope
+from netmover.logging_setup import get_logger
+from netmover.models import MountType, Rule, RunStatus
+from netmover.mounting.manager import MountManager, MountUnavailable
+from netmover.mounting.sftp import SftpError, connect
+from netmover.mover.destinations import (
+    Destination,
+    LocalDestination,
+    SftpDestination,
+    TransferError,
+)
+from netmover.mover.engine import Engine, RuleSnapshot, RunResult
+from netmover.mover.ledger import Ledger
+from netmover.timeouts import CancelToken, OperationTimeout, run_with_watchdog
+from netmover.validation import ValidationError, safe_abspath, safe_relpath
+
+log = get_logger(__name__)
+
+
+@dataclass(frozen=True)
+class RulePlan:
+    snapshot: RuleSnapshot
+    name: str
+    src_mount_id: int
+    src_type: MountType
+    src_subpath: str
+    dst_mount_id: int | None
+    dst_type: MountType | None
+    dst_name: str | None
+    dst_path: str
+    quarantine_path: str | None
+    op_timeout: int
+    dry_run: bool
+
+
+def parse_owner(value: str | None) -> tuple[int, int] | None:
+    if not value:
+        return None
+    uid, _, gid = value.partition(":")
+    try:
+        return int(uid), int(gid or uid)
+    except ValueError:
+        raise ValidationError("set_owner", "Eigentümer bitte als „uid:gid“ angeben") from None
+
+
+def parse_mode(value: str | None) -> int | None:
+    return int(value, 8) if value else None
+
+
+def load_plan(session: Session, rule_id: int) -> RulePlan:
+    r = session.get(Rule, rule_id)
+    if r is None:
+        raise LookupError(f"Regel {rule_id} existiert nicht")
+    dst = r.dst_mount
+    snap = RuleSnapshot(
+        id=r.id,
+        name=r.name,
+        action=r.action,
+        on_collision=r.on_collision,
+        pattern=r.pattern,
+        pattern_type=r.pattern_type,
+        recursive=r.recursive,
+        stability_seconds=r.stability_seconds,
+        min_age_seconds=r.min_age_seconds,
+        max_size_mb=r.max_size_mb,
+        set_owner=parse_owner(r.set_owner),
+        set_mode=parse_mode(r.set_mode),
+        quarantine_after=r.quarantine_after,
+        dst_root_squash=dst.root_squash if dst is not None else None,
+    )
+    return RulePlan(
+        snapshot=snap,
+        name=r.name,
+        src_mount_id=r.src_mount_id,
+        src_type=r.src_mount.type,
+        src_subpath=safe_relpath("src_subpath", r.src_subpath),
+        dst_mount_id=r.dst_mount_id,
+        dst_type=dst.type if dst is not None else None,
+        dst_name=dst.name if dst is not None else None,
+        dst_path=r.dst_path,
+        quarantine_path=safe_relpath("quarantine_path", r.quarantine_path)
+        if r.quarantine_path
+        else None,
+        op_timeout=r.op_timeout_seconds,
+        dry_run=r.dry_run,
+    )
+
+
+class RuleRunner:
+    def __init__(
+        self,
+        session_factory: sessionmaker[Session],
+        mounts: MountManager,
+        *,
+        sleep: Callable[[float], None] = time.sleep,
+        clock: Callable[[], float] = time.time,
+    ) -> None:
+        self.session_factory = session_factory
+        self.mounts = mounts
+        self.engine = Engine(Ledger(session_factory), sleep=sleep, clock=clock)
+        self._clock = clock
+        self._locks: dict[int, threading.Lock] = {}
+        self._started: dict[int, float] = {}
+        self._guard = threading.Lock()
+
+    def _lock(self, rule_id: int) -> threading.Lock:
+        with self._guard:
+            return self._locks.setdefault(rule_id, threading.Lock())
+
+    def running_since(self, rule_id: int) -> datetime | None:
+        ts = self._started.get(rule_id)
+        return datetime.fromtimestamp(ts, UTC) if ts is not None else None
+
+    def run(self, rule_id: int, *, dry_run: bool | None = None) -> RunResult:
+        lock = self._lock(rule_id)
+        if not lock.acquire(blocking=False):
+            since = self.running_since(rule_id)
+            when = f" (seit {since.astimezone():%H:%M:%S})" if since else ""
+            return RunResult(
+                status=RunStatus.SKIPPED,
+                dry_run=bool(dry_run),
+                message=f"Termin übersprungen – voriger Lauf läuft noch{when}",
+            )
+        # Wird True, sobald der Worker-Thread den Lock übernommen hat; ab dann gibt
+        # ihn ausschließlich der Worker frei (on_finish), auch nach einem Zeitlimit.
+        handed_over = [False]
+        try:
+            with session_scope(self.session_factory) as s:
+                plan = load_plan(s, rule_id)
+            dry = plan.dry_run if dry_run is None else dry_run
+            return self._run_locked(plan, dry, lock, rule_id, handed_over)
+        except (ValidationError, LookupError) as exc:
+            return RunResult(status=RunStatus.ERROR, dry_run=bool(dry_run), message=str(exc))
+        finally:
+            if not handed_over[0]:
+                lock.release()
+
+    def _run_locked(
+        self,
+        plan: RulePlan,
+        dry: bool,
+        lock: threading.Lock,
+        rule_id: int,
+        handed_over: list[bool],
+    ) -> RunResult:
+        if not plan.src_type.is_kernel_mount:
+            return RunResult(
+                RunStatus.ERROR, dry, message="SFTP-Anbindungen sind nur als Ziel möglich"
+            )
+        try:
+            src_mp = self.mounts.check_alive(plan.src_mount_id)
+            dst_mp = (
+                self.mounts.check_alive(plan.dst_mount_id)
+                if plan.dst_mount_id is not None
+                else None
+            )
+        except MountUnavailable as exc:
+            return RunResult(RunStatus.ERROR, dry, message=str(exc))
+        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
+        quarantine_root = src_mp / plan.quarantine_path if plan.quarantine_path else None
+
+        try:
+            make_dest = self._dest_factory(plan, dst_mp, src_root)
+        except (ValidationError, TransferError, SftpError) as exc:
+            return RunResult(RunStatus.ERROR, dry, message=str(exc))
+
+        def work(token: CancelToken) -> RunResult:
+            dest = make_dest()
+            try:
+                return self.engine.execute(
+                    plan.snapshot,
+                    src_root,
+                    dest,
+                    token,
+                    dry_run=dry,
+                    quarantine_root=quarantine_root,
+                )
+            finally:
+                dest.close()
+
+        def finish() -> None:
+            self._started.pop(rule_id, None)
+            lock.release()
+
+        self._started[rule_id] = self._clock()
+        handed_over[0] = True
+        try:
+            return run_with_watchdog(work, plan.op_timeout, f"rule:{plan.name}", on_finish=finish)
+        except OperationTimeout:
+            return RunResult(
+                RunStatus.ERROR,
+                dry,
+                message=(
+                    f"Keine Reaktion seit {plan.op_timeout} s – Lauf abgebrochen. Vermutlich hängt "
+                    "eine Anbindung; sie wird vor dem nächsten Lauf geprüft und ggf. neu "
+                    "eingehängt."
+                ),
+            )
+        except (TransferError, SftpError, OSError) as exc:
+            return RunResult(RunStatus.ERROR, dry, message=str(exc))
+
+    def _under_mount_base(self, path: Path) -> bool:
+        # Wörtlich und aufgelöst prüfen – ein Symlink darf die Prüfung nicht umgehen.
+        bases = {self.mounts.mount_base.absolute(), self.mounts.mount_base.resolve()}
+        paths = {path.absolute(), path.resolve()}
+        return any(p == b or b in p.parents for p in paths for b in bases)
+
+    def _dest_factory(
+        self, plan: RulePlan, dst_mp: Path | None, src_root: Path
+    ) -> Callable[[], Destination]:
+        if plan.dst_mount_id is None:
+            root = Path(safe_abspath("dst_path", plan.dst_path))
+            if self._under_mount_base(root):
+                base = self.mounts.mount_base
+                raise ValidationError(
+                    "dst_path",
+                    f"Lokales Ziel liegt unter {base} – bitte die Anbindung als Ziel wählen, "
+                    "damit geprüft wird, ob sie eingehängt ist.",
+                )
+            kind = "local"
+            local_root = root
+            create = False
+        elif plan.dst_type is MountType.SFTP:
+            scfg = self.mounts.sftp_config_for(plan.dst_mount_id)
+            sub = safe_relpath("dst_path", plan.dst_path)
+            remote = posixpath.join(scfg.base_path, sub) if sub else scfg.base_path
+            label = plan.dst_name or "sftp"
+
+            def make_sftp() -> Destination:
+                return SftpDestination(connect(scfg), remote, label=label)
+
+            return make_sftp
+        else:
+            if dst_mp is None:
+                raise TransferError("Ziel-Anbindung ist nicht eingehängt")
+            sub = safe_relpath("dst_path", plan.dst_path)
+            local_root = dst_mp / sub if sub else dst_mp
+            kind = "cifs" if plan.dst_type is MountType.CIFS else "nfs"
+            create = True
+
+        if local_root.resolve() == src_root.resolve():
+            raise ValidationError("dst_path", "Quelle und Ziel sind derselbe Ordner")
+
+        def make_local() -> Destination:
+            return LocalDestination(local_root, kind, create_root=create)  # type: ignore[arg-type]
+
+        return make_local

src/netmover/mover/scan.py

@@ -0,0 +1,110 @@
+"""Kandidaten im Quellordner finden (Fallstrick 1, erster Teil).
+
+Übersprungen werden immer:
+* versteckte Dateien und Ordner (führender Punkt) – darunter auch unsere eigenen
+  ``.netmover-*.part``-Temporärdateien,
+* typische Endungen unfertiger Übertragungen,
+* Symlinks und alles, was keine reguläre Datei ist,
+* Synology-Verwaltungsordner (``@eaDir``, ``#recycle``, ``#snapshot``).
+"""
+
+from __future__ import annotations
+
+import fnmatch
+import os
+import re
+import stat
+from collections.abc import Iterator
+from dataclasses import dataclass
+from pathlib import Path
+
+from netmover.models import PatternType
+
+PARTIAL_SUFFIXES = (".part", ".tmp", ".crdownload", ".filepart", ".!ut")
+IGNORED_DIRS = frozenset({"@eaDir", "#recycle", "#snapshot", "@tmp", "@sharebin"})
+
+
+@dataclass(frozen=True)
+class FileState:
+    size: int
+    mtime_ns: int
+
+
+@dataclass(frozen=True)
+class Candidate:
+    relpath: str  # POSIX, relativ zum Quellordner
+    path: Path
+    state: FileState
+
+
+def is_ignored_name(name: str) -> bool:
+    return name.startswith(".") or name.lower().endswith(PARTIAL_SUFFIXES)
+
+
+class Matcher:
+    """Glob: gegen den Dateinamen, ohne Groß-/Kleinschreibung, mehrere Muster mit
+    „;“ getrennt (``*.pdf;*.jpg``). Regex: ``re.search`` gegen den relativen Pfad."""
+
+    def __init__(self, pattern: str, pattern_type: PatternType) -> None:
+        self.pattern_type = pattern_type
+        if pattern_type is PatternType.REGEX:
+            self._regex = re.compile(pattern)
+            self._globs: list[str] = []
+        else:
+            self._globs = [p.strip().lower() for p in pattern.split(";") if p.strip()] or ["*"]
+
+    def __call__(self, relpath: str) -> bool:
+        if self.pattern_type is PatternType.REGEX:
+            return self._regex.search(relpath) is not None
+        name = relpath.rsplit("/", 1)[-1].lower()
+        return any(fnmatch.fnmatchcase(name, g) for g in self._globs)
+
+
+def validate_pattern(pattern: str, pattern_type: PatternType) -> None:
+    if pattern_type is PatternType.REGEX:
+        re.compile(pattern)
+
+
+def scan(
+    root: Path,
+    matcher: Matcher,
+    recursive: bool,
+    exclude: frozenset[Path] = frozenset(),
+) -> Iterator[Candidate]:
+    """Liefert passende reguläre Dateien. ``exclude``: Verzeichnisse (realpath), die
+    nicht betreten werden – z. B. Quarantäne oder ein Ziel innerhalb der Quelle."""
+
+    def walk(directory: Path, prefix: str) -> Iterator[Candidate]:
+        with os.scandir(directory) as it:
+            entries = sorted(it, key=lambda e: e.name)
+        for entry in entries:
+            if entry.name.startswith("."):
+                continue
+            rel = f"{prefix}{entry.name}"
+            if entry.is_dir(follow_symlinks=False):
+                if (
+                    recursive
+                    and entry.name not in IGNORED_DIRS
+                    and Path(entry.path).resolve() not in exclude
+                ):
+                    yield from walk(Path(entry.path), f"{rel}/")
+                continue
+            if is_ignored_name(entry.name):
+                continue
+            st = entry.stat(follow_symlinks=False)
+            if not stat.S_ISREG(st.st_mode):
+                continue
+            if matcher(rel):
+                yield Candidate(rel, Path(entry.path), FileState(st.st_size, st.st_mtime_ns))
+
+    yield from walk(root, "")
+
+
+def current_state(path: Path) -> FileState | None:
+    try:
+        st = os.stat(path, follow_symlinks=False)
+    except FileNotFoundError:
+        return None
+    if not stat.S_ISREG(st.st_mode):
+        return None
+    return FileState(st.st_size, st.st_mtime_ns)

src/netmover/timeouts.py

@@ -10,6 +10,7 @@ Regeln. Bewusst kein Thread-Pool – hängende Threads würden ihn auffüllen.
 from __future__ import annotations
 
 import threading
+import time
 from collections.abc import Callable
 
 from netmover.logging_setup import event, get_logger
@@ -56,3 +57,76 @@ def run_with_timeout[T](fn: Callable[[], T], seconds: float, what: str) -> T:
     if error:
         raise error[0]
     return result[0]
+
+
+class Cancelled(Exception):
+    """Lauf wurde abgebrochen (Zeitlimit oder Dienstende)."""
+
+
+class CancelToken:
+    """Abbruch-Flag plus Lebenszeichen für :func:`run_with_watchdog`.
+
+    Der Worker ruft :meth:`check` regelmäßig auf (pro Datei, pro Datenblock).
+    Das meldet Fortschritt und wirft :class:`Cancelled`, sobald abgebrochen wurde.
+    """
+
+    def __init__(self) -> None:
+        self._event = threading.Event()
+        self.last_beat = time.monotonic()
+
+    def beat(self) -> None:
+        self.last_beat = time.monotonic()
+
+    def cancel(self) -> None:
+        self._event.set()
+
+    @property
+    def cancelled(self) -> bool:
+        return self._event.is_set()
+
+    def check(self) -> None:
+        if self._event.is_set():
+            raise Cancelled
+        self.beat()
+
+
+def run_with_watchdog[T](
+    fn: Callable[[CancelToken], T],
+    idle_timeout: float,
+    what: str,
+    *,
+    on_finish: Callable[[], None] | None = None,
+    poll: float = 0.1,
+) -> T:
+    """Wie :func:`run_with_timeout`, aber das Zeitlimit gilt für *Inaktivität*:
+    Ein großer Transfer, der stetig Fortschritt meldet, wird nicht abgebrochen,
+    ein hängender Dateizugriff schon. ``on_finish`` läuft im Worker-Thread, wenn
+    dieser wirklich endet – auch lange nach einer Zeitüberschreitung."""
+    token = CancelToken()
+    result: list[T] = []
+    error: list[BaseException] = []
+    done = threading.Event()
+
+    def target() -> None:
+        try:
+            result.append(fn(token))
+        except BaseException as exc:
+            error.append(exc)
+        finally:
+            done.set()
+            if on_finish is not None:
+                on_finish()
+
+    thread = threading.Thread(target=target, name=f"nm-worker:{what}", daemon=True)
+    thread.start()
+    while not done.wait(poll):
+        if time.monotonic() - token.last_beat > idle_timeout:
+            token.cancel()
+            global _abandoned
+            with _abandoned_lock:
+                _abandoned += 1
+            event(log, "worker.timeout", level=30, what=what, idle_seconds=idle_timeout)
+            raise OperationTimeout(what, idle_timeout)
+    if error:
+        raise error[0]
+    return result[0]

tests/test_engine.py

@@ -0,0 +1,680 @@
+"""Mover-Engine mit lokalen Zielen – Fallstricke 1, 3, 4, 10, 11, 14 und Quarantäne."""
+
+from __future__ import annotations
+
+import errno
+import os
+import time
+from collections.abc import Callable
+from pathlib import Path
+
+import pytest
+from sqlalchemy import select
+from sqlalchemy.orm import Session, sessionmaker
+
+from netmover.db import session_scope
+from netmover.models import (
+    CollisionPolicy,
+    CopiedFile,
+    ItemStatus,
+    PatternType,
+    Rule,
+    RuleAction,
+    RunStatus,
+)
+from netmover.mover import destinations as dest_mod
+from netmover.mover.destinations import LocalDestination, TransferError
+from netmover.mover.engine import Engine, RuleSnapshot, RunResult
+from netmover.mover.ledger import Ledger
+from netmover.mover.naming import collision_candidates
+from netmover.mover.scan import Matcher
+from netmover.timeouts import CancelToken
+from tests.conftest import MountFactory
+
+
+@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="r", src_mount_id=mid, dst_path="/x")
+        s.add(r)
+        s.flush()
+        return r.id
+
+
+@pytest.fixture
+def ledger(session_factory: sessionmaker[Session]) -> Ledger:
+    return Ledger(session_factory)
+
+
+Sleeper = Callable[[float], None]
+
+
+@pytest.fixture
+def hooks() -> list[Callable[[], None]]:
+    """Aktionen, die während des Stabilitätsfensters ausgeführt werden."""
+    return []
+
+
+@pytest.fixture
+def engine(ledger: Ledger, hooks: list[Callable[[], None]]) -> Engine:
+    def sleep(_: float) -> None:
+        for h in hooks:
+            h()
+
+    return Engine(ledger, sleep=sleep)
+
+
+@pytest.fixture
+def cross_device(monkeypatch: pytest.MonkeyPatch) -> None:
+    """tmp_path liegt auf einem Gerät – für Kopier-Tests den Schnellweg abschalten."""
+    monkeypatch.setattr(LocalDestination, "move_same_device", lambda *a, **k: None)
+
+
+@pytest.fixture
+def src(tmp_path: Path) -> Path:
+    p = tmp_path / "src"
+    p.mkdir()
+    return p
+
+
+@pytest.fixture
+def dst(tmp_path: Path) -> Path:
+    p = tmp_path / "dst"
+    p.mkdir()
+    return p
+
+
+def snap(rule_id: int, **kw: object) -> RuleSnapshot:
+    return RuleSnapshot(id=rule_id, name="r", **kw)  # type: ignore[arg-type]
+
+
+def run(
+    engine: Engine, rule: RuleSnapshot, src: Path, dst: Path | LocalDestination, **kw: object
+) -> RunResult:
+    d = dst if isinstance(dst, LocalDestination) else LocalDestination(dst)
+    return engine.execute(rule, src, d, CancelToken(), **kw)  # type: ignore[arg-type]
+
+
+def write(path: Path, data: str = "inhalt") -> Path:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(data)
+    return path
+
+
+def statuses(result: RunResult) -> dict[str, ItemStatus]:
+    return {i.src: i.status for i in result.items}
+
+
+# -- Fallstrick 1: halb geschriebene Dateien --------------------------------------------
+
+
+def test_growing_file_is_not_touched(
+    engine: Engine, rule_id: int, src: Path, dst: Path, hooks: list[Callable[[], None]]
+) -> None:
+    write(src / "fertig.pdf", "a" * 100)
+    growing = write(src / "wachsend.pdf", "b" * 10)
+    hooks.append(lambda: growing.write_text("b" * 20))  # wächst während des Fensters
+    result = run(engine, snap(rule_id), src, dst)
+    assert statuses(result) == {"fertig.pdf": ItemStatus.MOVED, "wachsend.pdf": ItemStatus.SKIPPED}
+    assert growing.read_text() == "b" * 20
+    assert not (dst / "wachsend.pdf").exists()
+    assert (dst / "fertig.pdf").read_text() == "a" * 100
+
+
+def test_mtime_change_alone_counts_as_unstable(
+    engine: Engine, rule_id: int, src: Path, dst: Path, hooks: list[Callable[[], None]]
+) -> None:
+    f = write(src / "scan.pdf")
+    st = f.stat()
+    hooks.append(lambda: os.utime(f, ns=(st.st_atime_ns, st.st_mtime_ns + 1_000_000)))
+    result = run(engine, snap(rule_id), src, dst)
+    assert statuses(result) == {"scan.pdf": ItemStatus.SKIPPED}
+    assert f.exists()
+
+
+def test_stability_window_is_waited(ledger: Ledger, rule_id: int, src: Path, dst: Path) -> None:
+    waited: list[float] = []
+    eng = Engine(ledger, sleep=waited.append)
+    write(src / "a.pdf")
+    run(eng, snap(rule_id, stability_seconds=7), src, dst)
+    assert waited == [7]
+    waited.clear()
+    run(eng, snap(rule_id, stability_seconds=7), src, dst)  # nichts mehr da → kein Warten
+    assert waited == []
+
+
+@pytest.mark.parametrize(
+    "name",
+    [".versteckt.pdf", "scan.pdf.part", "x.tmp", "dl.crdownload", "f.filepart", "t.!ut", "A.PART"],
+)
+def test_hidden_and_partial_files_are_ignored(
+    engine: Engine, rule_id: int, src: Path, dst: Path, name: str
+) -> None:
+    write(src / name)
+    result = run(engine, snap(rule_id), src, dst)
+    assert result.seen == 0
+    assert (src / name).exists()
+
+
+def test_min_age_and_max_size(engine: Engine, rule_id: int, src: Path, dst: Path) -> None:
+    write(src / "neu.pdf")
+    old = write(src / "alt.pdf")
+    os.utime(old, (time.time() - 3600, time.time() - 3600))
+    write(src / "gross.pdf", "x" * (2 * 1024 * 1024))
+    os.utime(src / "gross.pdf", (time.time() - 3600, time.time() - 3600))
+    result = run(engine, snap(rule_id, min_age_seconds=600, max_size_mb=1), src, dst)
+    assert statuses(result) == {
+        "alt.pdf": ItemStatus.MOVED,
+        "gross.pdf": ItemStatus.SKIPPED,
+        "neu.pdf": ItemStatus.SKIPPED,
+    }
+
+
+# -- Fallstrick 3: Atomarität ---------------------------------------------------------------
+
+
+@pytest.mark.usefixtures("cross_device")
+def test_final_name_never_shows_partial_content(
+    engine: Engine, rule_id: int, src: Path, dst: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    data = "x" * (3 * dest_mod.CHUNK + 17)
+    write(src / "gross.pdf", data)
+    seen_during_write: list[list[str]] = []
+    original = LocalDestination._upload
+
+    def spy(self: LocalDestination, *a: object, **kw: object) -> list[str]:
+        notes = original(self, *a, **kw)  # type: ignore[arg-type]
+        seen_during_write.append(sorted(p.name for p in dst.iterdir()))
+        return notes
+
+    monkeypatch.setattr(LocalDestination, "_upload", spy)
+    result = run(engine, snap(rule_id), src, dst)
+    assert statuses(result) == {"gross.pdf": ItemStatus.MOVED}
+    # während/nach dem Schreiben existiert nur die .part-Datei, nicht der Endname
+    (names,) = seen_during_write
+    assert len(names) == 1 and names[0].startswith(".netmover-") and names[0].endswith(".part")
+    assert (dst / "gross.pdf").read_text() == data
+    assert not (src / "gross.pdf").exists()
+
+
+STAGES = ["_upload", "_apply_meta", "_link_noreplace", "verify"]
+
+
+@pytest.mark.usefixtures("cross_device")
+@pytest.mark.parametrize("stage", STAGES)
+def test_abort_at_any_stage_keeps_source_and_leaves_no_part(
+    engine: Engine, rule_id: int, src: Path, dst: Path, monkeypatch: pytest.MonkeyPatch, stage: str
+) -> None:
+    write(src / "scan.pdf", "wichtig")
+    if stage == "verify":
+        # Quelle wird zwischen Kopie und Umbenennen verändert
+        original = LocalDestination._apply_meta
+
+        def touch(self: LocalDestination, rel: str, meta: object) -> list[str]:
+            (src / "scan.pdf").write_text("wichtig!!")
+            return original(self, rel, meta)  # type: ignore[arg-type]
+
+        monkeypatch.setattr(LocalDestination, "_apply_meta", touch)
+    else:
+
+        def boom(self: object, *a: object, **kw: object) -> None:
+            raise OSError(errno.EIO if stage == "_upload" else errno.ENOSPC, "Fehler")
+
+        monkeypatch.setattr(LocalDestination, stage, boom)
+    result = run(engine, snap(rule_id), src, dst)
+    assert (src / "scan.pdf").exists()
+    assert list(dst.iterdir()) == []
+    item = result.items[0] if result.items else None
+    if stage == "_upload":  # EIO → Anbindung gestört, ganzer Lauf abgebrochen
+        assert result.status is RunStatus.ERROR
+    elif stage == "verify":
+        assert item is not None and item.status is ItemStatus.SKIPPED
+    else:
+        assert item is not None and item.status is ItemStatus.FAILED
+
+
+@pytest.mark.parametrize("path", ["same-device", "copy"])
+def test_source_not_deletable_keeps_target_and_prevents_duplicate(
+    engine: Engine, rule_id: int, src: Path, dst: Path, monkeypatch: pytest.MonkeyPatch, path: str
+) -> None:
+    if path == "copy":
+        monkeypatch.setattr(LocalDestination, "move_same_device", lambda *a, **k: None)
+    write(src / "scan.pdf")
+    real_unlink = os.unlink
+
+    def deny(path: object, *a: object, **kw: object) -> None:
+        if str(path).startswith(str(src)):
+            raise PermissionError(errno.EACCES, "Permission denied", str(path))
+        real_unlink(path)  # type: ignore[arg-type]
+
+    monkeypatch.setattr("netmover.mover.engine.os.unlink", deny)
+    first = run(engine, snap(rule_id), src, dst)
+    assert first.items[0].status is ItemStatus.FAILED
+    assert "nicht löschbar" in (first.items[0].message or "")
+    assert (dst / "scan.pdf").exists() and (src / "scan.pdf").exists()
+    second = run(engine, snap(rule_id), src, dst)
+    assert second.items == [] and second.message == "1 bereits übertragen"
+    assert sorted(p.name for p in dst.iterdir()) == ["scan.pdf"]
+
+
+def test_stale_part_files_are_cleaned(engine: Engine, rule_id: int, src: Path, dst: Path) -> None:
+    old = write(dst / ".netmover-dead.part", "rest")
+    young = write(dst / ".netmover-busy.part", "rest")
+    os.utime(old, (time.time() - 7200, time.time() - 7200))
+    run(engine, snap(rule_id), src, dst)
+    assert not old.exists()
+    assert young.exists()
+
+
+def test_same_device_move_uses_link_without_copy(
+    engine: Engine, rule_id: int, src: Path, dst: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    write(src / "a.pdf", "x")
+    ino = (src / "a.pdf").stat().st_ino
+
+    def no_copy(*a: object, **kw: object) -> None:
+        raise AssertionError("darf auf demselben Gerät nicht kopieren")
+
+    monkeypatch.setattr(LocalDestination, "_upload", no_copy)
+    result = run(engine, snap(rule_id), src, dst)
+    assert statuses(result) == {"a.pdf": ItemStatus.MOVED}
+    assert (dst / "a.pdf").stat().st_ino == ino
+    assert not (src / "a.pdf").exists()
+
+
+def test_cross_device_path_is_used_when_devices_differ(
+    engine: Engine, rule_id: int, src: Path, dst: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.setattr(LocalDestination, "move_same_device", lambda *a, **k: None)
+    write(src / "a.pdf", "x")
+    ino = (src / "a.pdf").stat().st_ino
+    result = run(engine, snap(rule_id), src, dst)
+    assert statuses(result) == {"a.pdf": ItemStatus.MOVED}
+    assert (dst / "a.pdf").stat().st_ino != ino
+
+
+def test_filesystem_without_hardlinks_falls_back_to_rename(
+    engine: Engine, rule_id: int, src: Path, dst: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.setattr(LocalDestination, "move_same_device", lambda *a, **k: None)
+
+    def no_link(*a: object, **kw: object) -> None:
+        raise OSError(errno.EPERM, "Operation not permitted")
+
+    monkeypatch.setattr("netmover.mover.destinations.os.link", no_link)
+    write(src / "a.pdf")
+    write(dst / "a.pdf", "alt")
+    result = run(engine, snap(rule_id), src, dst)
+    assert result.items[0].dst == str(dst / "a-1.pdf")
+    assert (dst / "a.pdf").read_text() == "alt"
+
+
+# -- Fallstrick 4: Namenskollisionen ---------------------------------------------------------
+
+
+def test_collision_candidates() -> None:
+    gen = collision_candidates("sub/scan001.pdf")
+    assert [next(gen) for _ in range(3)] == [
+        "sub/scan001.pdf",
+        "sub/scan001-1.pdf",
+        "sub/scan001-2.pdf",
+    ]
+    gen = collision_candidates("README")
+    assert [next(gen) for _ in range(2)] == ["README", "README-1"]
+
+
+def test_rename_counts_up_against_filesystem(
+    engine: Engine, rule_id: int, src: Path, dst: Path
+) -> None:
+    write(dst / "scan001.pdf", "alt")
+    for i in range(3):
+        write(src / "scan001.pdf", f"neu{i}")
+        run(engine, snap(rule_id), src, dst)
+    assert sorted(p.name for p in dst.iterdir()) == [
+        "scan001-1.pdf",
+        "scan001-2.pdf",
+        "scan001-3.pdf",
+        "scan001.pdf",
+    ]
+    assert (dst / "scan001.pdf").read_text() == "alt"
+    assert (dst / "scan001-3.pdf").read_text() == "neu2"
+
+
+def test_rename_rechecks_filesystem_each_time(
+    engine: Engine, rule_id: int, src: Path, dst: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """Zwischen Plan und Umbenennen belegt ein Dritter den Namen – kein Überschreiben."""
+    monkeypatch.setattr(LocalDestination, "move_same_device", lambda *a, **k: None)
+    write(dst / "scan.pdf", "alt")
+    write(src / "scan.pdf", "neu")
+    original = LocalDestination._apply_meta
+
+    def intruder(self: LocalDestination, rel: str, meta: object) -> list[str]:
+        write(dst / "scan-1.pdf", "fremd")
+        return original(self, rel, meta)  # type: ignore[arg-type]
+
+    monkeypatch.setattr(LocalDestination, "_apply_meta", intruder)
+    result = run(engine, snap(rule_id), src, dst)
+    assert result.items[0].dst == str(dst / "scan-2.pdf")
+    assert (dst / "scan-1.pdf").read_text() == "fremd"
+    assert (dst / "scan-2.pdf").read_text() == "neu"
+
+
+def test_skip_policy(engine: Engine, rule_id: int, src: Path, dst: Path) -> None:
+    write(dst / "a.pdf", "alt")
+    write(src / "a.pdf", "neu")
+    result = run(engine, snap(rule_id, on_collision=CollisionPolicy.SKIP), src, dst)
+    assert statuses(result) == {"a.pdf": ItemStatus.SKIPPED}
+    assert (dst / "a.pdf").read_text() == "alt"
+    assert (src / "a.pdf").exists()
+
+
+def test_overwrite_policy(
+    engine: Engine, rule_id: int, src: Path, dst: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.setattr(LocalDestination, "move_same_device", lambda *a, **k: None)
+    write(dst / "a.pdf", "alt")
+    write(src / "a.pdf", "neu")
+    result = run(engine, snap(rule_id, on_collision=CollisionPolicy.OVERWRITE), src, dst)
+    assert statuses(result) == {"a.pdf": ItemStatus.MOVED}
+    assert (dst / "a.pdf").read_text() == "neu"
+
+
+# -- Fallstrick 11: Kopiermodus mit Gedächtnis ---------------------------------------------
+
+
+def test_copy_mode_copies_each_version_once(
+    engine: Engine, rule_id: int, src: Path, dst: Path, session_factory: sessionmaker[Session]
+) -> None:
+    rule = snap(rule_id, action=RuleAction.COPY)
+    f = write(src / "scan.pdf", "v1")
+    for _ in range(3):
+        run(engine, rule, src, dst)
+    assert sorted(p.name for p in dst.iterdir()) == ["scan.pdf"]
+    assert f.exists()
+
+    # neue Version → läuft durch die Kollisionsregel
+    f.write_text("version 2")
+    os.utime(f, ns=(f.stat().st_atime_ns, f.stat().st_mtime_ns + 5_000_000_000))
+    result = run(engine, rule, src, dst)
+    assert statuses(result) == {"scan.pdf": ItemStatus.COPIED}
+    run(engine, rule, src, dst)
+    assert sorted(p.name for p in dst.iterdir()) == ["scan-1.pdf", "scan.pdf"]
+    assert (dst / "scan-1.pdf").read_text() == "version 2"
+
+    # Quelle verschwindet → Buchführung wird bereinigt
+    f.unlink()
+    run(engine, rule, src, dst)
+    with session_factory() as s:
+        assert s.scalars(select(CopiedFile)).all() == []
+
+
+def test_copy_mode_preserves_mtime(engine: Engine, rule_id: int, src: Path, dst: Path) -> None:
+    f = write(src / "a.pdf")
+    os.utime(f, ns=(1_600_000_000_000_000_000, 1_600_000_000_123_456_789))
+    run(engine, snap(rule_id, action=RuleAction.COPY), src, dst)
+    assert (dst / "a.pdf").stat().st_mtime_ns == 1_600_000_000_123_456_789
+
+
+# -- Fallstrick 14: Platz ---------------------------------------------------------------------
+
+
+def test_insufficient_space_skips_file_without_partial_copy(
+    engine: Engine, rule_id: int, src: Path, dst: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    write(src / "a.pdf", "x" * 1000)
+    monkeypatch.setattr(LocalDestination, "free_bytes", lambda self: 1040)  # < 1000 * 1.05
+    result = run(engine, snap(rule_id), src, dst)
+    assert result.items[0].status is ItemStatus.FAILED
+    assert "zu wenig Platz" in (result.items[0].message or "")
+    assert list(dst.iterdir()) == []
+    assert (src / "a.pdf").exists()
+
+
+def test_metadata_failure_is_only_a_hint(
+    engine: Engine, rule_id: int, src: Path, dst: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.setattr(LocalDestination, "move_same_device", lambda *a, **k: None)
+
+    def deny(*a: object, **kw: object) -> None:
+        raise PermissionError(errno.EPERM, "Operation not permitted")
+
+    monkeypatch.setattr("netmover.mover.destinations.os.utime", deny)
+    write(src / "a.pdf")
+    result = run(engine, snap(rule_id), src, dst)
+    assert result.items[0].status is ItemStatus.MOVED
+    assert "Änderungszeit nicht übernommen" in (result.items[0].message or "")
+
+
+# -- Fallstrick 10: Berechtigungen ---------------------------------------------------------
+
+
+def test_set_mode_applied(engine: Engine, rule_id: int, src: Path, dst: Path) -> None:
+    write(src / "a.pdf").chmod(0o600)
+    run(engine, snap(rule_id, set_mode=0o640), src, dst)
+    assert (dst / "a.pdf").stat().st_mode & 0o777 == 0o640
+
+
+def test_unchanged_mode_is_copied_from_source(
+    engine: Engine, rule_id: int, src: Path, dst: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.setattr(LocalDestination, "move_same_device", lambda *a, **k: None)
+    write(src / "a.pdf").chmod(0o604)
+    run(engine, snap(rule_id), src, dst)
+    assert (dst / "a.pdf").stat().st_mode & 0o777 == 0o604
+
+
+def test_chown_without_permission_is_clear_message(
+    engine: Engine, rule_id: int, src: Path, dst: Path
+) -> None:
+    if os.geteuid() == 0:
+        pytest.skip("als root ist chown erlaubt")
+    write(src / "a.pdf")
+    result = run(engine, snap(rule_id, set_owner=(0, 0)), src, dst)
+    assert result.items[0].status is ItemStatus.MOVED
+    assert "keine Berechtigung" in (result.items[0].message or "")
+
+
+@pytest.mark.parametrize(("kind", "squash"), [("cifs", None), ("nfs", True), ("nfs", None)])
+def test_chown_skipped_on_cifs_and_squashed_nfs_warns_once(
+    engine: Engine,
+    rule_id: int,
+    src: Path,
+    dst: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    kind: str,
+    squash: bool | None,
+) -> None:
+    called: list[object] = []
+    monkeypatch.setattr("netmover.mover.destinations.os.chown", lambda *a: called.append(a))
+    write(src / "a.pdf")
+    write(src / "b.pdf")
+    d = LocalDestination(dst, kind)  # type: ignore[arg-type]
+    result = run(engine, snap(rule_id, set_owner=(1000, 1000), dst_root_squash=squash), src, d)
+    assert called == []
+    msgs = [i.message for i in result.items if i.message and "chown übersprungen" in i.message]
+    assert len(msgs) == 1  # einmalig warnen
+
+
+def test_chown_on_nfs_without_squash_is_tried(
+    engine: Engine, rule_id: int, src: Path, dst: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    called: list[object] = []
+    monkeypatch.setattr("netmover.mover.destinations.os.chown", lambda *a: called.append(a))
+    write(src / "a.pdf")
+    run(
+        engine,
+        snap(rule_id, set_owner=(1000, 100), dst_root_squash=False),
+        src,
+        LocalDestination(dst, "nfs"),
+    )
+    assert called and called[0][1:] == (1000, 100)  # type: ignore[index]
+
+
+# -- Filter, Rekursion, Probelauf, Quarantäne ---------------------------------------------
+
+
+def test_matcher() -> None:
+    m = Matcher("*.pdf; *.JPG", PatternType.GLOB)
+    assert m("a/Scan.PDF") and m("b.jpg") and not m("c.txt")
+    r = Matcher(r"^rechnungen/.*\.pdf$", PatternType.REGEX)
+    assert r("rechnungen/2026.pdf") and not r("scan/2026.pdf")
+
+
+def test_recursive_preserves_structure_and_skips_synology_dirs(
+    engine: Engine, rule_id: int, src: Path, dst: Path
+) -> None:
+    write(src / "max" / "a.pdf")
+    write(src / "@eaDir" / "a.pdf" / "SYNOFILE_THUMB_M.jpg")
+    write(src / "#recycle" / "alt.pdf")
+    write(src / "top.pdf")
+    flat = run(engine, snap(rule_id), src, dst)
+    assert statuses(flat) == {"top.pdf": ItemStatus.MOVED}
+    rec = run(engine, snap(rule_id, recursive=True), src, dst)
+    assert statuses(rec) == {"max/a.pdf": ItemStatus.MOVED}
+    assert (dst / "max" / "a.pdf").exists()
+    assert (src / "#recycle" / "alt.pdf").exists()
+
+
+def test_symlinks_are_ignored(
+    engine: Engine, rule_id: int, src: Path, dst: Path, tmp_path: Path
+) -> None:
+    secret = write(tmp_path / "geheim.txt")
+    (src / "link.txt").symlink_to(secret)
+    result = run(engine, snap(rule_id), src, dst)
+    assert result.seen == 0 and secret.exists()
+
+
+def test_dry_run_touches_nothing(
+    engine: Engine, rule_id: int, src: Path, dst: Path, session_factory: sessionmaker[Session]
+) -> None:
+    write(src / "a.pdf")
+    write(dst / "a.pdf", "alt")
+    write(src / "b.pdf")
+    result = run(engine, snap(rule_id, action=RuleAction.COPY), src, dst, dry_run=True)
+    assert {i.src: (i.status, i.dst) for i in result.items} == {
+        "a.pdf": (ItemStatus.PLANNED, str(dst / "a-1.pdf")),
+        "b.pdf": (ItemStatus.PLANNED, str(dst / "b.pdf")),
+    }
+    assert sorted(p.name for p in dst.iterdir()) == ["a.pdf"]
+    assert sorted(p.name for p in src.iterdir()) == ["a.pdf", "b.pdf"]
+    with session_factory() as s:
+        assert s.scalars(select(CopiedFile)).all() == []
+
+
+def test_quarantine_after_repeated_failures(
+    engine: Engine,
+    rule_id: int,
+    src: Path,
+    dst: Path,
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setattr(LocalDestination, "move_same_device", lambda *a, **k: None)
+
+    def broken(self: object, *a: object, **kw: object) -> None:
+        raise TransferError("kaputt")
+
+    monkeypatch.setattr(LocalDestination, "_link_noreplace", broken)
+    write(src / "defekt.pdf")
+    quarantine = src / "quarantaene"
+    rule = snap(rule_id, quarantine_after=3, recursive=True)
+    msgs = []
+    for _ in range(3):
+        r = run(engine, rule, src, dst, quarantine_root=quarantine)
+        msgs.append(r.items[0].message)
+    assert msgs[-1] == "kaputt (Versuch 3)"
+    r = run(engine, rule, src, dst, quarantine_root=quarantine)
+    assert statuses(r) == {"defekt.pdf": ItemStatus.QUARANTINED}
+    assert (quarantine / "defekt.pdf").exists()
+    # Quarantäne liegt in der Quelle, wird aber trotz Rekursion nicht erneut gescannt
+    r = run(engine, rule, src, dst, quarantine_root=quarantine)
+    assert r.seen == 0
+
+
+def test_mount_error_mid_run_aborts_instead_of_failing_every_file(
+    engine: Engine, rule_id: int, src: Path, dst: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.setattr(LocalDestination, "move_same_device", lambda *a, **k: None)
+
+    def stale(self: object, *a: object, **kw: object) -> None:
+        raise OSError(errno.ESTALE, "Stale file handle")
+
+    monkeypatch.setattr(LocalDestination, "_upload", stale)
+    write(src / "a.pdf")
+    write(src / "b.pdf")
+    result = run(engine, snap(rule_id), src, dst)
+    assert result.status is RunStatus.ERROR
+    assert "gestört" in (result.message or "")
+    assert (src / "a.pdf").exists() and (src / "b.pdf").exists()
+
+
+def test_cancel_stops_before_next_file(engine: Engine, rule_id: int, src: Path, dst: Path) -> None:
+    write(src / "a.pdf")
+    token = CancelToken()
+    token.cancel()
+    result = engine.execute(snap(rule_id), src, LocalDestination(dst), token)
+    assert result.status is RunStatus.ERROR
+    assert (src / "a.pdf").exists()
+
+
+def test_missing_source_dir(engine: Engine, rule_id: int, tmp_path: Path, dst: Path) -> None:
+    result = run(engine, snap(rule_id), tmp_path / "fehlt", dst)
+    assert result.status is RunStatus.ERROR and "existiert nicht" in (result.message or "")
+
+
+def test_local_destination_must_exist(
+    engine: Engine, rule_id: int, src: Path, tmp_path: Path
+) -> None:
+    write(src / "a.pdf")
+    result = run(engine, snap(rule_id), src, LocalDestination(tmp_path / "fehlt"))
+    assert result.status is RunStatus.ERROR
+    assert (src / "a.pdf").exists()
+
+
+@pytest.mark.usefixtures("cross_device")
+def test_fsync_before_final_rename(
+    engine: Engine, rule_id: int, src: Path, dst: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    events: list[str] = []
+    real_fsync, real_link = os.fsync, os.link
+
+    def fsync(fd: int) -> None:
+        events.append("fsync")
+        real_fsync(fd)
+
+    def link(a: object, b: object, *args: object, **kw: object) -> None:
+        events.append(f"link:{Path(str(b)).name}")
+        real_link(a, b)  # type: ignore[arg-type]
+
+    monkeypatch.setattr("netmover.mover.destinations.os.fsync", fsync)
+    monkeypatch.setattr("netmover.mover.destinations.os.link", link)
+    write(src / "a.pdf")
+    run(engine, snap(rule_id), src, dst)
+    assert events.index("fsync") < events.index("link:a.pdf")
+
+
+@pytest.mark.usefixtures("cross_device")
+def test_name_taken_between_check_and_link_is_never_overwritten(
+    engine: Engine, rule_id: int, src: Path, dst: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """Enges Wettlauf-Fenster: exists() sagt „frei“, link() meldet EEXIST."""
+    write(dst / "scan.pdf", "fremd")
+    write(src / "scan.pdf", "neu")
+    real_exists = LocalDestination.exists
+    lied: list[str] = []
+
+    def stale_exists(self: LocalDestination, rel: str) -> bool:
+        if rel == "scan.pdf" and not lied and self.root == dst:
+            lied.append(rel)
+            return False  # veraltete Antwort
+        return real_exists(self, rel)
+
+    monkeypatch.setattr(LocalDestination, "plan", lambda self, rel, policy: rel)
+    monkeypatch.setattr(LocalDestination, "exists", stale_exists)
+    result = run(engine, snap(rule_id), src, dst)
+    assert lied
+    assert result.items[0].dst == str(dst / "scan-1.pdf")
+    assert (dst / "scan.pdf").read_text() == "fremd"
+    assert (dst / "scan-1.pdf").read_text() == "neu"

tests/test_runner.py

@@ -0,0 +1,307 @@
+"""Regellauf mit FakeMountBackend – Fallstricke 2, 8 und 13 sowie SFTP-Ziel."""
+
+from __future__ import annotations
+
+import threading
+import time
+from collections.abc import Iterator
+from pathlib import Path
+
+import pytest
+from sqlalchemy.orm import Session, sessionmaker
+
+from netmover.crypto import SecretBox
+from netmover.db import session_scope
+from netmover.models import (
+    Credential,
+    CredentialKind,
+    ItemStatus,
+    Mount,
+    MountType,
+    Rule,
+    RunStatus,
+)
+from netmover.mounting.fake import FakeMountBackend
+from netmover.mounting.manager import MountManager
+from netmover.mover.runner import RuleRunner
+from tests.conftest import MountFactory
+from tests.sftp_stub import StubSFTPServer
+
+
+@pytest.fixture
+def setup(
+    manager: MountManager,
+    fake_backend: FakeMountBackend,
+    make_mount: MountFactory,
+    session_factory: sessionmaker[Session],
+) -> tuple[int, int, int, Path, Path]:
+    scans = fake_backend.export("nas", "scans")
+    archiv = fake_backend.export("nas2", "archiv")
+    src_id = make_mount("scans", server="nas", share="scans")
+    dst_id = make_mount("archiv", server="nas2", share="archiv")
+    manager.mount(src_id)
+    manager.mount(dst_id)
+    with session_scope(session_factory) as s:
+        r = Rule(
+            name="scans→archiv",
+            src_mount_id=src_id,
+            dst_mount_id=dst_id,
+            dst_path="eingang",
+            stability_seconds=1,
+        )
+        s.add(r)
+        s.flush()
+        rid = r.id
+    return rid, src_id, dst_id, scans, archiv
+
+
+@pytest.fixture
+def runner(session_factory: sessionmaker[Session], manager: MountManager) -> RuleRunner:
+    return RuleRunner(session_factory, manager, sleep=lambda _: None)
+
+
+def test_moves_between_two_mounts(
+    runner: RuleRunner, setup: tuple[int, int, int, Path, Path]
+) -> None:
+    rid, _, _, scans, archiv = setup
+    (scans / "scan001.pdf").write_text("x")
+    result = runner.run(rid)
+    assert result.status is RunStatus.OK
+    assert [(i.src, i.status) for i in result.items] == [("scan001.pdf", ItemStatus.MOVED)]
+    assert (archiv / "eingang" / "scan001.pdf").read_text() == "x"
+    assert not (scans / "scan001.pdf").exists()
+
+
+# -- Fallstrick 2 ---------------------------------------------------------------------------
+
+
+def test_unmounted_destination_aborts_and_writes_nothing_locally(
+    runner: RuleRunner,
+    setup: tuple[int, int, int, Path, Path],
+    fake_backend: FakeMountBackend,
+    manager: MountManager,
+) -> None:
+    rid, _, _, scans, _archiv = setup
+    (scans / "scan001.pdf").write_text("x")
+    mp = manager.mountpoint_for("archiv")
+    fake_backend.vanish(mp)  # Freigabe weg, Mountpoint ist jetzt ein leeres lokales Verzeichnis
+    result = runner.run(rid)
+    assert result.status is RunStatus.ERROR
+    assert "nicht eingehängt" in (result.message or "")
+    assert list(mp.iterdir()) == []  # nichts auf die lokale Platte geschrieben
+    assert (scans / "scan001.pdf").exists()
+
+
+def test_unmounted_source_aborts(
+    runner: RuleRunner,
+    setup: tuple[int, int, int, Path, Path],
+    fake_backend: FakeMountBackend,
+    manager: MountManager,
+) -> None:
+    rid, *_ = setup
+    fake_backend.vanish(manager.mountpoint_for("scans"))
+    result = runner.run(rid)
+    assert result.status is RunStatus.ERROR and result.items == []
+
+
+def test_local_destination_under_mount_base_is_refused(
+    runner: RuleRunner,
+    setup: tuple[int, int, int, Path, Path],
+    manager: MountManager,
+    session_factory: sessionmaker[Session],
+) -> None:
+    rid, *_ = setup
+    with session_scope(session_factory) as s:
+        r = s.get(Rule, rid)
+        assert r is not None
+        r.dst_mount_id = None
+        r.dst_path = str(manager.mountpoint_for("archiv"))
+    result = runner.run(rid)
+    assert result.status is RunStatus.ERROR
+    assert "Anbindung als Ziel" in (result.message or "")
+
+
+def test_local_destination(
+    runner: RuleRunner,
+    setup: tuple[int, int, int, Path, Path],
+    tmp_path: Path,
+    session_factory: sessionmaker[Session],
+) -> None:
+    rid, _, _, scans, _ = setup
+    consume = tmp_path / "consume"
+    consume.mkdir()
+    with session_scope(session_factory) as s:
+        r = s.get(Rule, rid)
+        assert r is not None
+        r.dst_mount_id = None
+        r.dst_path = str(consume)
+    (scans / "a.pdf").write_text("x")
+    assert runner.run(rid).status is RunStatus.OK
+    assert (consume / "a.pdf").exists()
+
+
+# -- Fallstrick 8 / 13: Überlappung und Zeitlimit ---------------------------------------------
+
+
+@pytest.fixture
+def blocking_runner(
+    session_factory: sessionmaker[Session], manager: MountManager
+) -> Iterator[tuple[RuleRunner, threading.Event, threading.Event]]:
+    entered = threading.Event()
+    release = threading.Event()
+
+    def sleep(_: float) -> None:  # Stabilitätsfenster „hängt“
+        entered.set()
+        release.wait(10)
+
+    yield RuleRunner(session_factory, manager, sleep=sleep), entered, release
+    release.set()
+
+
+def test_overlapping_run_is_skipped(
+    blocking_runner: tuple[RuleRunner, threading.Event, threading.Event],
+    setup: tuple[int, int, int, Path, Path],
+) -> None:
+    runner, entered, release = blocking_runner
+    rid, _, _, scans, archiv = setup
+    (scans / "a.pdf").write_text("x")
+    results = []
+    t = threading.Thread(target=lambda: results.append(runner.run(rid)))
+    t.start()
+    assert entered.wait(5)
+    skipped = runner.run(rid)
+    assert skipped.status is RunStatus.SKIPPED
+    assert "läuft noch" in (skipped.message or "")
+    release.set()
+    t.join(5)
+    assert results[0].status is RunStatus.OK
+    assert (archiv / "eingang" / "a.pdf").exists()
+
+
+def test_hanging_run_times_out_and_blocks_rule_until_worker_ends(
+    blocking_runner: tuple[RuleRunner, threading.Event, threading.Event],
+    setup: tuple[int, int, int, Path, Path],
+    session_factory: sessionmaker[Session],
+) -> None:
+    runner, _entered, release = blocking_runner
+    rid, _, _, scans, archiv = setup
+    with session_scope(session_factory) as s:
+        r = s.get(Rule, rid)
+        assert r is not None
+        r.op_timeout_seconds = 1
+    (scans / "a.pdf").write_text("x")
+
+    t0 = time.monotonic()
+    result = runner.run(rid)
+    assert time.monotonic() - t0 < 5  # Scheduler wird nicht blockiert
+    assert result.status is RunStatus.ERROR
+    assert "Keine Reaktion" in (result.message or "")
+    assert runner.running_since(rid) is not None
+
+    # solange der alte Worker hängt: neue Termine überspringen
+    assert runner.run(rid).status is RunStatus.SKIPPED
+
+    # Worker wacht auf → sieht den Abbruch, verschiebt nichts mehr, gibt Lock frei
+    release.set()
+    for _ in range(50):
+        if runner.running_since(rid) is None:
+            break
+        time.sleep(0.05)
+    assert runner.running_since(rid) is None
+    assert (scans / "a.pdf").exists()
+    assert not (archiv / "eingang" / "a.pdf").exists()
+
+
+def test_stale_source_is_recovered_before_run(
+    runner: RuleRunner,
+    setup: tuple[int, int, int, Path, Path],
+    fake_backend: FakeMountBackend,
+    manager: MountManager,
+) -> None:
+    import errno
+
+    rid, _, _, scans, archiv = setup
+    (scans / "a.pdf").write_text("x")
+    fake_backend.inject_fault(manager.mountpoint_for("scans"), errno.ESTALE)
+    result = runner.run(rid)
+    assert result.status is RunStatus.OK
+    assert (archiv / "eingang" / "a.pdf").exists()
+    assert ("unmount", True) in [(c.action, c.lazy) for c in fake_backend.calls]
+
+
+def test_dry_run_flag_from_rule_and_override(
+    runner: RuleRunner,
+    setup: tuple[int, int, int, Path, Path],
+    session_factory: sessionmaker[Session],
+) -> None:
+    rid, _, _, scans, _archiv = setup
+    (scans / "a.pdf").write_text("x")
+    with session_scope(session_factory) as s:
+        r = s.get(Rule, rid)
+        assert r is not None
+        r.dry_run = True
+    result = runner.run(rid)
+    assert result.dry_run and result.items[0].status is ItemStatus.PLANNED
+    assert (scans / "a.pdf").exists()
+    assert runner.run(rid, dry_run=False).items[0].status is ItemStatus.MOVED
+
+
+def test_unknown_rule(runner: RuleRunner) -> None:
+    assert runner.run(999).status is RunStatus.ERROR
+
+
+# -- SFTP-Ziel --------------------------------------------------------------------------------
+
+
+@pytest.fixture
+def sftp_server(tmp_path: Path) -> Iterator[StubSFTPServer]:
+    root = tmp_path / "sftp-root"
+    (root / "consume").mkdir(parents=True)
+    srv = StubSFTPServer(root).start()
+    yield srv
+    srv.stop()
+
+
+def test_move_to_sftp_destination_with_collision(
+    runner: RuleRunner,
+    setup: tuple[int, int, int, Path, Path],
+    sftp_server: StubSFTPServer,
+    session_factory: sessionmaker[Session],
+    secretbox: SecretBox,
+) -> None:
+    rid, _, _, scans, _ = setup
+    with session_scope(session_factory) as s:
+        c = Credential(
+            name="pk",
+            kind=CredentialKind.SSH_KEY,
+            username=sftp_server.username,
+            secret_enc=secretbox.encrypt(sftp_server.client_private_key),
+        )
+        s.add(c)
+        s.flush()
+        m = Mount(
+            name="paperless",
+            slug="paperless",
+            type=MountType.SFTP,
+            server="127.0.0.1",
+            share="/consume",
+            port=sftp_server.port,
+            cred_id=c.id,
+            host_key=sftp_server.host_key_string,
+        )
+        s.add(m)
+        s.flush()
+        r = s.get(Rule, rid)
+        assert r is not None
+        r.dst_mount_id = m.id
+        r.dst_path = ""
+    consume = sftp_server.root / "consume"
+    (consume / "scan.pdf").write_text("alt")
+    (scans / "scan.pdf").write_bytes(b"%PDF neu" * 50_000)
+    result = runner.run(rid)
+    assert result.status is RunStatus.OK, result.message
+    assert result.items[0].dst == "paperless:/consume/scan-1.pdf"
+    assert (consume / "scan-1.pdf").read_bytes() == b"%PDF neu" * 50_000
+    assert (consume / "scan.pdf").read_text() == "alt"
+    assert not (scans / "scan.pdf").exists()
+    assert sorted(p.name for p in consume.iterdir()) == ["scan-1.pdf", "scan.pdf"]