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/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

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"]