feat: Archiv je Regel – erst Archiv, dann Ziel, dann Quelle löschen

Zusätzliche Kopie ins Archiv (Anbindung oder lokaler Pfad, Ablage wie Quelle
oder nach Datum). Buchführung verhindert doppelte Archivkopien bei Neuversuch;
im Archiv wird nie überschrieben. Migration 0004 mit Tests auf befüllten
Datenbanken (inkl. Downgrade).

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

src/netmover/migrations/versions/0004_archive.py

@@ -0,0 +1,103 @@
+"""archive
+
+Revision ID: 0004
+Revises: 0003
+Create Date: 2026-10-12
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+import netmover.models
+
+revision: str = "0004"
+down_revision: str | None = "0003"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "archived_files",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("rule_id", sa.Integer(), nullable=False),
+        sa.Column("src_relpath", sa.String(length=2048), nullable=False),
+        sa.Column("size", sa.BigInteger(), nullable=False),
+        sa.Column("mtime_ns", sa.BigInteger(), nullable=False),
+        sa.Column("archive_path", sa.String(length=2048), nullable=False),
+        sa.Column("archived_at", netmover.models.UTCDateTime(), nullable=False),
+        sa.ForeignKeyConstraint(
+            ["rule_id"],
+            ["rules.id"],
+            name=op.f("fk_archived_files_rule_id_rules"),
+            ondelete="CASCADE",
+        ),
+        sa.PrimaryKeyConstraint("id", name=op.f("pk_archived_files")),
+        sa.UniqueConstraint(
+            "rule_id",
+            "src_relpath",
+            "size",
+            "mtime_ns",
+            name=op.f("uq_archived_files_rule_id_src_relpath_size_mtime_ns"),
+        ),
+    )
+    with op.batch_alter_table("rules", schema=None) as batch_op:
+        batch_op.add_column(
+            sa.Column(
+                "archive_kind",
+                sa.Enum(
+                    "none",
+                    "mount",
+                    "local",
+                    name="archive_kind",
+                    native_enum=False,
+                    create_constraint=True,
+                    length=24,
+                ),
+                server_default="none",
+                nullable=False,
+            )
+        )
+        batch_op.add_column(sa.Column("archive_mount_id", sa.Integer(), nullable=True))
+        batch_op.add_column(
+            sa.Column("archive_path", sa.String(length=1024), server_default="", nullable=False)
+        )
+        batch_op.add_column(
+            sa.Column(
+                "archive_layout",
+                sa.Enum(
+                    "flat",
+                    "date",
+                    name="archive_layout",
+                    native_enum=False,
+                    create_constraint=True,
+                    length=24,
+                ),
+                server_default="flat",
+                nullable=False,
+            )
+        )
+        batch_op.create_foreign_key(
+            batch_op.f("fk_rules_archive_mount_id_mounts"),
+            "mounts",
+            ["archive_mount_id"],
+            ["id"],
+            ondelete="RESTRICT",
+        )
+
+
+def downgrade() -> None:
+    with op.batch_alter_table("rules", schema=None) as batch_op:
+        batch_op.drop_constraint(batch_op.f("fk_rules_archive_mount_id_mounts"), type_="foreignkey")
+        # CHECK-Regeln der Auswahlfelder zuerst entfernen – sonst verweisen sie beim
+        # Neuaufbau der Tabelle auf Spalten, die es nicht mehr gibt.
+        batch_op.drop_constraint(batch_op.f("ck_rules_archive_kind"), type_="check")
+        batch_op.drop_constraint(batch_op.f("ck_rules_archive_layout"), type_="check")
+        batch_op.drop_column("archive_layout")
+        batch_op.drop_column("archive_path")
+        batch_op.drop_column("archive_mount_id")
+        batch_op.drop_column("archive_kind")
+
+    op.drop_table("archived_files")

src/netmover/models.py

@@ -10,6 +10,7 @@ Abweichungen bzw. Ergänzungen gegenüber dem Mindestumfang aus der Spezifikatio
 * ``credentials.managed`` + ``mounts.username`` (Migration 0002) –
   Schlüsselverwaltung: gespeicherte SSH-Schlüssel, die mehrere Anbindungen mit
   jeweils eigenem Benutzernamen nutzen.
+* ``rules.archive_*`` + Tabelle ``archived_files`` (Migration 0004) – Archivkopie.
 * ``rules.waiting_since`` / ``rules.waiting_reason`` (Migration 0003) – Regel wartet
   auf eine nicht erreichbare Anbindung (automatisch pausiert).
 * ``mounts.state_since`` – für „hängt seit …“ / „gestört seit …“ auf der Statusseite.
@@ -138,6 +139,17 @@ class CollisionPolicy(enum.StrEnum):
     OVERWRITE = "overwrite"
 
 
+class ArchiveKind(enum.StrEnum):
+    NONE = "none"
+    MOUNT = "mount"
+    LOCAL = "local"
+
+
+class ArchiveLayout(enum.StrEnum):
+    FLAT = "flat"  # wie in der Quelle
+    DATE = "date"  # Jahr/Monat/…, nach Tag der Verarbeitung
+
+
 class RunStatus(enum.StrEnum):
     RUNNING = "running"
     OK = "ok"  # keine Dateifehler
@@ -259,6 +271,18 @@ class Rule(Base):
     quarantine_path: Mapped[str | None] = mapped_column(String(1024))
     quarantine_after: Mapped[int] = mapped_column(Integer, default=3)
 
+    # Archiv (Migration 0004): zusätzliche Kopie; Quelle erst löschen, wenn Ziel UND Archiv da
+    archive_kind: Mapped[ArchiveKind] = mapped_column(
+        _enum(ArchiveKind, "archive_kind"), default=ArchiveKind.NONE, server_default="none"
+    )
+    archive_mount_id: Mapped[int | None] = mapped_column(
+        ForeignKey("mounts.id", ondelete="RESTRICT")
+    )
+    archive_path: Mapped[str] = mapped_column(String(1024), default="", server_default="")
+    archive_layout: Mapped[ArchiveLayout] = mapped_column(
+        _enum(ArchiveLayout, "archive_layout"), default=ArchiveLayout.FLAT, server_default="flat"
+    )
+
     last_run_at: Mapped[datetime | None] = mapped_column(UTCDateTime)
     last_success_at: Mapped[datetime | None] = mapped_column(UTCDateTime)
     consecutive_failures: Mapped[int] = mapped_column(Integer, default=0)
@@ -269,6 +293,7 @@ class Rule(Base):
 
     src_mount: Mapped[Mount] = relationship(foreign_keys=[src_mount_id])
     dst_mount: Mapped[Mount | None] = relationship(foreign_keys=[dst_mount_id])
+    archive_mount: Mapped[Mount | None] = relationship(foreign_keys=[archive_mount_id])
     runs: Mapped[list[Run]] = relationship(
         back_populates="rule", cascade="all, delete-orphan", passive_deletes=True
     )
@@ -330,6 +355,22 @@ class CopiedFile(Base):
     copied_at: Mapped[datetime] = mapped_column(UTCDateTime, default=utcnow)
 
 
+class ArchivedFile(Base):
+    """Welche Dateiversion schon im Archiv liegt – damit ein Neuversuch nach einem
+    gescheiterten Ziel keine zweite Archivkopie anlegt."""
+
+    __tablename__ = "archived_files"
+    __table_args__ = (UniqueConstraint("rule_id", "src_relpath", "size", "mtime_ns"),)
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True)
+    rule_id: Mapped[int] = mapped_column(ForeignKey("rules.id", ondelete="CASCADE"))
+    src_relpath: Mapped[str] = mapped_column(String(2048))
+    size: Mapped[int] = mapped_column(BigInteger)
+    mtime_ns: Mapped[int] = mapped_column(BigInteger)
+    archive_path: Mapped[str] = mapped_column(String(2048))
+    archived_at: Mapped[datetime] = mapped_column(UTCDateTime, default=utcnow)
+
+
 class FileFailure(Base):
     """Fehlversuche pro Datei, Grundlage für die Quarantäne."""
 

src/netmover/mover/engine.py

@@ -19,6 +19,7 @@ import posixpath
 import time
 from collections.abc import Callable
 from dataclasses import dataclass, field
+from datetime import datetime
 from pathlib import Path
 from typing import Any
 
@@ -72,6 +73,7 @@ class RuleSnapshot:
     set_mode: int | None = None
     quarantine_after: int = 3
     dst_root_squash: bool | None = None  # nur NFS-Ziele: Ergebnis des Schreibtests
+    archive_layout: str = "flat"  # "flat" | "date" (Jahr/Monat/…)
 
 
 @dataclass
@@ -159,6 +161,7 @@ class Engine:
         dry_run: bool = False,
         quarantine_rel: str | None = None,
         quarantine_root: Path | None = None,
+        archive: Destination | None = None,
     ) -> RunResult:
         """``source``: eine :class:`Source` oder (Kurzform) ein lokales Verzeichnis.
         Quarantäne relativ zur Wurzel der Anbindung (``quarantine_rel``) oder – bei
@@ -168,7 +171,7 @@ class Engine:
         if quarantine_root is not None and isinstance(src, LocalSource):
             quarantine_rel = str(quarantine_root.relative_to(src.root))
         try:
-            self._execute(rule, src, dest, token, result, dry_run, quarantine_rel)
+            self._execute(rule, src, dest, token, result, dry_run, quarantine_rel, archive)
         except Cancelled:
             result.status = RunStatus.ERROR
             result.message = "Lauf abgebrochen"
@@ -195,10 +198,13 @@ class Engine:
         result: RunResult,
         dry_run: bool,
         quarantine_rel: str | None,
+        archive: Destination | None = None,
     ) -> None:
         src.check_root()
         if not dry_run:
             dest.prepare()
+            if archive is not None:
+                archive.prepare()
         token.check()
 
         exclude = frozenset({quarantine_rel}) if quarantine_rel else frozenset()
@@ -212,8 +218,12 @@ class Engine:
         # 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)
+        archived = self.ledger.archived_keys(rule.id) if archive is not None else set()
         if not dry_run:
-            self.ledger.prune_copied(rule.id, {c.relpath for c in candidates})
+            present = {c.relpath for c in candidates}
+            self.ledger.prune_copied(rule.id, present)
+            if archive is not None:
+                self.ledger.prune_archived(rule.id, present)
         fresh = [c for c in candidates if (c.relpath, c.state.size, c.state.mtime_ns) not in done]
         unchanged = len(candidates) - len(fresh)
 
@@ -271,7 +281,11 @@ class Engine:
         failures = self.ledger.failures(rule.id)
         for c in stable:
             token.check()
-            result.add(self._process(rule, c, src, dest, token, dry_run, quarantine_rel, failures))
+            result.add(
+                self._process(
+                    rule, c, src, dest, token, dry_run, quarantine_rel, failures, archive, archived
+                )
+            )
 
         if unchanged:
             result.message = f"{unchanged} bereits übertragen"
@@ -309,8 +323,14 @@ class Engine:
         dry_run: bool,
         quarantine_rel: str | None,
         failures: dict[str, FailureInfo],
+        archive: Destination | None = None,
+        archived: set[tuple[str, int, int]] | None = None,
     ) -> ItemResult:
         size = c.state.size
+        arch_rel = self._archive_rel(c.relpath, rule.archive_layout)
+        need_archive = archive is not None and (
+            (c.relpath, size, c.state.mtime_ns) not in (archived or set())
+        )
         failure = failures.get(c.relpath)
 
         # Quarantäne für wiederholt scheiternde Dateien
@@ -349,9 +369,14 @@ class Engine:
                 else None
             )
             verb = "kopieren" if rule.action is RuleAction.COPY else "verschieben"
-            return ItemResult(
+            msg = msg or f"würde {verb}"
-                c.relpath, ItemStatus.PLANNED, dest.describe(planned), size, msg or f"würde {verb}"
+            if archive is not None:
-            )
+                if need_archive:
+                    arch_planned = archive.plan(arch_rel, CollisionPolicy.RENAME)
+                    msg += f"; Archiv: {archive.describe(arch_planned or arch_rel)}"
+                else:
+                    msg += "; bereits archiviert"
+            return ItemResult(c.relpath, ItemStatus.PLANNED, dest.describe(planned), size, msg)
 
         try:
             free = dest.free_bytes()
@@ -367,10 +392,51 @@ class Engine:
                 f"benötigt {_mb(int(size * SPACE_RESERVE))})",
             )
 
+        if need_archive and archive is not None:
+            try:
+                afree = archive.free_bytes()
+            except OSError:
+                afree = None
+            if afree is not None and afree < size * SPACE_RESERVE:
+                return ItemResult(
+                    c.relpath,
+                    ItemStatus.FAILED,
+                    archive.describe(arch_rel),
+                    size,
+                    f"zu wenig Platz im Archiv (frei {_mb(afree)}, "
+                    f"benötigt {_mb(int(size * SPACE_RESERVE))})",
+                )
+
         def verify() -> None:
             if src.state(c.relpath) != c.state:
                 raise SourceChanged("Quelle hat sich während der Übertragung geändert")
 
+        # Archiv ZUERST: scheitert es, ist noch nichts ans Ziel (z. B. Paperless)
+        # geliefert. Scheitert danach das Ziel, verhindert die Buchführung eine
+        # zweite Archivkopie beim nächsten Versuch.
+        archive_note: str | None = None
+        if need_archive and archive is not None:
+            try:
+                meta, _ = self._meta(rule, src, c.relpath, dest)
+                ares = archive.commit(
+                    _SourceFile(src, c.relpath),
+                    size,
+                    arch_rel,
+                    CollisionPolicy.RENAME,
+                    MetaOptions(meta.atime_ns, meta.mtime_ns, meta.mode, None),
+                    verify,
+                    token,
+                )
+            except CollisionSkip:  # kann bei RENAME nicht auftreten
+                return ItemResult(c.relpath, ItemStatus.SKIPPED, None, size, "Archiv belegt")
+            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, prefix="Archiv: ")
+            archive_path = archive.describe(ares.final_rel)
+            self.ledger.record_archive(rule.id, c.relpath, size, c.state.mtime_ns, archive_path)
+            archive_note = f"archiviert: {archive_path}"
+
         try:
             meta, notes = self._meta(rule, src, c.relpath, dest)
             res = None
@@ -401,6 +467,8 @@ class Engine:
             return self._fail(rule, c, exc, dry_run)
 
         notes += res.notes
+        if archive_note:
+            notes.insert(0, archive_note)
         target = dest.describe(res.final_rel)
         if c.relpath in failures:
             self.ledger.clear_failure(rule.id, c.relpath)
@@ -425,10 +493,12 @@ class Engine:
                 )
         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:
+    def _fail(
+        self, rule: RuleSnapshot, c: Candidate, exc: Exception, dry_run: bool, prefix: str = ""
+    ) -> ItemResult:
         if isinstance(exc, OSError) and exc.errno in MOUNT_ERRNOS:
-            raise RunAborted(f"Anbindung gestört: {_describe_os_error(exc)}") from exc
+            raise RunAborted(f"{prefix}Anbindung gestört: {_describe_os_error(exc)}") from exc
-        msg = _describe_os_error(exc) if isinstance(exc, OSError) else str(exc)
+        msg = prefix + (_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
@@ -437,6 +507,12 @@ class Engine:
         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 _archive_rel(self, relpath: str, layout: str) -> str:
+        if layout == "date":
+            t = datetime.fromtimestamp(self._clock())
+            return f"{t:%Y}/{t:%m}/{relpath}"
+        return relpath
+
     def _quarantine(
         self, rule: RuleSnapshot, c: Candidate, src: Source, q_rel: str, attempts: int
     ) -> ItemResult:

src/netmover/mover/ledger.py

@@ -14,7 +14,7 @@ from sqlalchemy import delete, select
 from sqlalchemy.orm import Session, sessionmaker
 
 from netmover.db import session_scope
-from netmover.models import CopiedFile, FileFailure, utcnow
+from netmover.models import ArchivedFile, CopiedFile, FileFailure, utcnow
 
 CopyKey = tuple[str, int, int]  # (relpath, size, mtime_ns)
 
@@ -72,6 +72,53 @@ class Ledger:
                 s.execute(delete(CopiedFile).where(CopiedFile.id.in_(stale)))
             return len(stale)
 
+    # -- Archiv ------------------------------------------------------------------------
+
+    def archived_keys(self, rule_id: int) -> set[CopyKey]:
+        with session_scope(self.session_factory) as s:
+            rows = s.execute(
+                select(ArchivedFile.src_relpath, ArchivedFile.size, ArchivedFile.mtime_ns).where(
+                    ArchivedFile.rule_id == rule_id
+                )
+            ).all()
+        return {(r[0], r[1], r[2]) for r in rows}
+
+    def record_archive(
+        self, rule_id: int, relpath: str, size: int, mtime_ns: int, archive_path: str
+    ) -> None:
+        with session_scope(self.session_factory) as s:
+            exists = s.scalar(
+                select(ArchivedFile.id).where(
+                    ArchivedFile.rule_id == rule_id,
+                    ArchivedFile.src_relpath == relpath,
+                    ArchivedFile.size == size,
+                    ArchivedFile.mtime_ns == mtime_ns,
+                )
+            )
+            if exists is None:
+                s.add(
+                    ArchivedFile(
+                        rule_id=rule_id,
+                        src_relpath=relpath,
+                        size=size,
+                        mtime_ns=mtime_ns,
+                        archive_path=archive_path,
+                    )
+                )
+
+    def prune_archived(self, rule_id: int, present: set[str]) -> int:
+        """Wie :meth:`prune_copied`: nur nach einem vollständigen Scan aufrufen."""
+        with session_scope(self.session_factory) as s:
+            rows = s.execute(
+                select(ArchivedFile.id, ArchivedFile.src_relpath).where(
+                    ArchivedFile.rule_id == rule_id
+                )
+            ).all()
+            stale = [r[0] for r in rows if r[1] not in present]
+            if stale:
+                s.execute(delete(ArchivedFile).where(ArchivedFile.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()

src/netmover/mover/runner.py

@@ -56,6 +56,11 @@ class RulePlan:
     quarantine_path: str | None
     op_timeout: int
     dry_run: bool
+    archive_kind: str = "none"
+    archive_mount_id: int | None = None
+    archive_type: MountType | None = None
+    archive_name: str | None = None
+    archive_path: str = ""
 
 
 def parse_owner(value: str | None) -> tuple[int, int] | None:
@@ -92,6 +97,7 @@ def load_plan(session: Session, rule_id: int) -> RulePlan:
         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,
+        archive_layout=r.archive_layout.value,
     )
     return RulePlan(
         snapshot=snap,
@@ -109,6 +115,11 @@ def load_plan(session: Session, rule_id: int) -> RulePlan:
         else None,
         op_timeout=r.op_timeout_seconds,
         dry_run=r.dry_run,
+        archive_kind=r.archive_kind.value,
+        archive_mount_id=r.archive_mount_id if r.archive_kind.value == "mount" else None,
+        archive_type=r.archive_mount.type if r.archive_mount is not None else None,
+        archive_name=r.archive_mount.name if r.archive_mount is not None else None,
+        archive_path=r.archive_path,
     )
 
 
@@ -176,24 +187,52 @@ class RuleRunner:
                 if plan.dst_mount_id is not None
                 else None
             )
+            arch_mp = (
+                self.mounts.check_alive(plan.archive_mount_id)
+                if plan.archive_mount_id is not None
+                else None
+            )
         except MountUnavailable as exc:
             return RunResult(RunStatus.ERROR, dry, message=str(exc), unavailable=True)
 
         try:
             make_source = self._source_factory(plan, src_mp)
             src_root = src_mp / plan.src_subpath if src_mp and plan.src_subpath else src_mp
-            make_dest = self._dest_factory(plan, dst_mp, src_root)
+            make_dest = self._target(
+                "dst_path",
+                plan.dst_mount_id,
+                plan.dst_type,
+                plan.dst_name,
+                plan.dst_path,
+                dst_mp,
+                src_root,
+            )
+            make_archive = (
+                self._target(
+                    "archive_path",
+                    plan.archive_mount_id,
+                    plan.archive_type,
+                    plan.archive_name,
+                    plan.archive_path,
+                    arch_mp,
+                    src_root,
+                )
+                if plan.archive_kind != "none"
+                else None
+            )
         except (ValidationError, TransferError, SftpError) as exc:
             return RunResult(RunStatus.ERROR, dry, message=str(exc))
 
         def work(token: CancelToken) -> RunResult:
-            source = make_source()
+            opened: list[Source | Destination] = []
             try:
+                source = make_source()
+                opened.append(source)
                 dest = make_dest()
-            except BaseException:
+                opened.append(dest)
-                source.close()
+                archive = make_archive() if make_archive is not None else None
-                raise
+                if archive is not None:
-            try:
+                    opened.append(archive)
                 return self.engine.execute(
                     plan.snapshot,
                     source,
@@ -201,10 +240,11 @@ class RuleRunner:
                     token,
                     dry_run=dry,
                     quarantine_rel=plan.quarantine_path,
+                    archive=archive,
                 )
             finally:
-                dest.close()
+                for item in reversed(opened):
-                source.close()
+                    item.close()
 
         def finish() -> None:
             self._started.pop(rule_id, None)
@@ -251,41 +291,50 @@ class RuleRunner:
 
         return make_local
 
-    def _dest_factory(
+    def _target(
-        self, plan: RulePlan, dst_mp: Path | None, src_root: Path | None
+        self,
+        field: str,
+        mount_id: int | None,
+        mount_type: MountType | None,
+        mount_name: str | None,
+        path: str,
+        mp: Path | None,
+        src_root: Path | None,
     ) -> Callable[[], Destination]:
-        if plan.dst_mount_id is None:
+        """Ziel oder Archiv: lokaler Pfad (``mount_id`` None), SFTP oder Kernel-Mount."""
-            root = Path(safe_abspath("dst_path", plan.dst_path))
+        what = "Archiv" if field == "archive_path" else "Ziel"
+        if mount_id is None:
+            root = Path(safe_abspath(field, path))
             if self._under_mount_base(root):
                 base = self.mounts.mount_base
                 raise ValidationError(
-                    "dst_path",
+                    field,
-                    f"Lokales Ziel liegt unter {base} – bitte die Anbindung als Ziel wählen, "
+                    f"Lokales {what} liegt unter {base} – bitte die Anbindung als {what} 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:
+        elif mount_type is MountType.SFTP:
-            scfg = self.mounts.sftp_config_for(plan.dst_mount_id)
+            scfg = self.mounts.sftp_config_for(mount_id)
-            sub = safe_relpath("dst_path", plan.dst_path)
+            sub = safe_relpath(field, path)
             remote = posixpath.join(scfg.base_path, sub) if sub else scfg.base_path
-            label = plan.dst_name or "sftp"
+            label = mount_name or "sftp"
 
             def make_sftp() -> Destination:
                 return SftpDestination(connect(scfg), remote, label=label)
 
             return make_sftp
         else:
-            if dst_mp is None:
+            if mp is None:
-                raise TransferError("Ziel-Anbindung ist nicht eingehängt")
+                raise TransferError(f"{what}-Anbindung ist nicht eingehängt")
-            sub = safe_relpath("dst_path", plan.dst_path)
+            sub = safe_relpath(field, path)
-            local_root = dst_mp / sub if sub else dst_mp
+            local_root = mp / sub if sub else mp
-            kind = "cifs" if plan.dst_type is MountType.CIFS else "nfs"
+            kind = "cifs" if mount_type is MountType.CIFS else "nfs"
             create = True
 
         if src_root is not None and local_root.resolve() == src_root.resolve():
-            raise ValidationError("dst_path", "Quelle und Ziel sind derselbe Ordner")
+            raise ValidationError(field, f"Quelle und {what} sind derselbe Ordner")
 
         def make_local() -> Destination:
             return LocalDestination(local_root, kind, create_root=create)  # type: ignore[arg-type]

tests/test_archive.py

@@ -0,0 +1,351 @@
+"""0.7.0: Archiv – Kopie ans Ziel UND ins Archiv, Quelle erst danach löschen."""
+
+from __future__ import annotations
+
+import os
+from collections.abc import Iterator
+from datetime import datetime
+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 (
+    ArchivedFile,
+    ArchiveKind,
+    CollisionPolicy,
+    ItemStatus,
+    Rule,
+    RuleAction,
+    RunStatus,
+)
+from netmover.mounting.fake import FakeMountBackend
+from netmover.mounting.manager import MountManager
+from netmover.mover.destinations import LocalDestination, TransferError
+from netmover.mover.engine import Engine, RuleSnapshot, RunResult
+from netmover.mover.ledger import Ledger
+from netmover.mover.runner import RuleRunner
+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
+
+
+FIXED = datetime(2026, 10, 12, 9, 30).timestamp()
+
+
+@pytest.fixture
+def engine(session_factory: sessionmaker[Session]) -> Engine:
+    return Engine(Ledger(session_factory), sleep=lambda _: None, clock=lambda: FIXED)
+
+
+@pytest.fixture
+def dirs(tmp_path: Path) -> tuple[Path, Path, Path]:
+    src, dst, arch = tmp_path / "src", tmp_path / "dst", tmp_path / "archiv"
+    for d in (src, dst, arch):
+        d.mkdir()
+    return src, dst, arch
+
+
+@pytest.fixture(autouse=True)
+def cross_device(monkeypatch: pytest.MonkeyPatch) -> None:
+    """tmp_path liegt auf einem Gerät – realistischer Weg über Kopie erzwingen."""
+    monkeypatch.setattr(LocalDestination, "move_same_device", lambda *a, **k: None)
+
+
+def snap(rule_id: int, **kw: object) -> RuleSnapshot:
+    return RuleSnapshot(id=rule_id, name="r", stability_seconds=0, **kw)  # type: ignore[arg-type]
+
+
+def run(
+    engine: Engine, rule: RuleSnapshot, src: Path, dst: Path, arch: Path | None, **kw: object
+) -> RunResult:
+    archive = LocalDestination(arch) if arch is not None else None
+    return engine.execute(rule, src, LocalDestination(dst), CancelToken(), archive=archive, **kw)  # type: ignore[arg-type]
+
+
+def write(p: Path, data: bytes = b"inhalt") -> Path:
+    p.parent.mkdir(parents=True, exist_ok=True)
+    p.write_bytes(data)
+    return p
+
+
+def names(d: Path) -> list[str]:
+    return sorted(str(p.relative_to(d)) for p in d.rglob("*") if p.is_file())
+
+
+def test_move_delivers_to_target_and_archive_then_deletes_source(
+    engine: Engine, rule_id: int, dirs: tuple[Path, Path, Path]
+) -> None:
+    src, dst, arch = dirs
+    data = os.urandom(200_000)
+    write(src / "scan001.pdf", data)
+    result = run(engine, snap(rule_id), src, dst, arch)
+    assert result.status is RunStatus.OK
+    item = result.items[0]
+    assert item.status is ItemStatus.MOVED and item.message and "archiviert:" in item.message
+    assert (dst / "scan001.pdf").read_bytes() == data == (arch / "scan001.pdf").read_bytes()
+    assert not (src / "scan001.pdf").exists()
+    assert names(dst) == ["scan001.pdf"] and names(arch) == ["scan001.pdf"]  # keine .part-Reste
+
+
+def test_date_layout(engine: Engine, rule_id: int, dirs: tuple[Path, Path, Path]) -> None:
+    src, dst, arch = dirs
+    write(src / "max" / "a.pdf")
+    run(engine, snap(rule_id, recursive=True, archive_layout="date"), src, dst, arch)
+    assert names(arch) == ["2026/10/max/a.pdf"] and names(dst) == ["max/a.pdf"]
+
+
+def test_archive_failure_delivers_nothing_and_keeps_source(
+    engine: Engine, rule_id: int, dirs: tuple[Path, Path, Path], monkeypatch: pytest.MonkeyPatch
+) -> None:
+    src, dst, arch = dirs
+    write(src / "a.pdf")
+    real = LocalDestination._link_noreplace
+
+    def fail_in_archive(self: LocalDestination, tmp: str, final: str) -> None:
+        if self.root == arch:
+            raise TransferError("Archiv voll")
+        real(self, tmp, final)
+
+    monkeypatch.setattr(LocalDestination, "_link_noreplace", fail_in_archive)
+    result = run(engine, snap(rule_id), src, dst, arch)
+    item = result.items[0]
+    assert item.status is ItemStatus.FAILED and item.message == "Archiv: Archiv voll (Versuch 1)"
+    assert (src / "a.pdf").exists() and names(dst) == [] and names(arch) == []
+
+
+def test_target_failure_after_archive_never_archives_twice(
+    engine: Engine,
+    rule_id: int,
+    dirs: tuple[Path, Path, Path],
+    monkeypatch: pytest.MonkeyPatch,
+    session_factory: sessionmaker[Session],
+) -> None:
+    src, dst, arch = dirs
+    write(src / "a.pdf")
+    real = LocalDestination._link_noreplace
+
+    def fail_in_target(self: LocalDestination, tmp: str, final: str) -> None:
+        if self.root == dst:
+            raise TransferError("Ziel gesperrt")
+        real(self, tmp, final)
+
+    monkeypatch.setattr(LocalDestination, "_link_noreplace", fail_in_target)
+    first = run(engine, snap(rule_id), src, dst, arch)
+    assert first.items[0].status is ItemStatus.FAILED
+    assert (src / "a.pdf").exists() and names(arch) == ["a.pdf"] and names(dst) == []
+    with session_factory() as s:
+        assert len(s.scalars(select(ArchivedFile)).all()) == 1
+
+    monkeypatch.setattr(LocalDestination, "_link_noreplace", real)  # Ziel wieder frei
+    second = run(engine, snap(rule_id), src, dst, arch)
+    assert second.items[0].status is ItemStatus.MOVED
+    assert names(arch) == ["a.pdf"]  # keine zweite Archivkopie (a-1.pdf)
+    assert names(dst) == ["a.pdf"] and not (src / "a.pdf").exists()
+
+
+def test_archive_never_overwrites(
+    engine: Engine, rule_id: int, dirs: tuple[Path, Path, Path]
+) -> None:
+    src, dst, arch = dirs
+    write(arch / "scan.pdf", b"alt im Archiv")
+    write(src / "scan.pdf", b"neu")
+    run(engine, snap(rule_id, on_collision=CollisionPolicy.OVERWRITE), src, dst, arch)
+    assert (arch / "scan.pdf").read_bytes() == b"alt im Archiv"
+    assert (arch / "scan-1.pdf").read_bytes() == b"neu"
+
+
+def test_no_space_in_archive_writes_nothing(
+    engine: Engine, rule_id: int, dirs: tuple[Path, Path, Path], monkeypatch: pytest.MonkeyPatch
+) -> None:
+    src, dst, arch = dirs
+    write(src / "a.pdf", b"x" * 1000)
+    real = LocalDestination.free_bytes
+    monkeypatch.setattr(
+        LocalDestination, "free_bytes", lambda self: 10 if self.root == arch else real(self)
+    )
+    result = run(engine, snap(rule_id), src, dst, arch)
+    assert "zu wenig Platz im Archiv" in (result.items[0].message or "")
+    assert names(dst) == [] and names(arch) == [] and (src / "a.pdf").exists()
+
+
+def test_source_changed_during_archive_copy(
+    engine: Engine, rule_id: int, dirs: tuple[Path, Path, Path], monkeypatch: pytest.MonkeyPatch
+) -> None:
+    src, dst, arch = dirs
+    f = write(src / "a.pdf", b"original")
+    real = LocalDestination._apply_meta
+
+    def change(self: LocalDestination, rel: str, meta: object) -> list[str]:
+        if self.root == arch:
+            f.write_bytes(b"geaendert!!")
+        return real(self, rel, meta)  # type: ignore[arg-type]
+
+    monkeypatch.setattr(LocalDestination, "_apply_meta", change)
+    result = run(engine, snap(rule_id), src, dst, arch)
+    assert result.items[0].status is ItemStatus.SKIPPED
+    assert names(dst) == [] and names(arch) == [] and f.exists()
+
+
+def test_copy_mode_with_archive(
+    engine: Engine, rule_id: int, dirs: tuple[Path, Path, Path]
+) -> None:
+    src, dst, arch = dirs
+    f = write(src / "a.pdf", b"v1")
+    rule = snap(rule_id, action=RuleAction.COPY)
+    for _ in range(3):
+        run(engine, rule, src, dst, arch)
+    assert f.exists() and names(dst) == ["a.pdf"] and names(arch) == ["a.pdf"]
+    f.write_bytes(b"version 2")
+    run(engine, rule, src, dst, arch)
+    assert names(dst) == ["a-1.pdf", "a.pdf"] and names(arch) == ["a-1.pdf", "a.pdf"]
+
+
+def test_dry_run_shows_archive_and_writes_nothing(
+    engine: Engine, rule_id: int, dirs: tuple[Path, Path, Path]
+) -> None:
+    src, dst, arch = dirs
+    write(src / "a.pdf")
+    result = run(engine, snap(rule_id, archive_layout="date"), src, dst, arch, dry_run=True)
+    assert result.items[0].status is ItemStatus.PLANNED
+    assert f"Archiv: {arch / '2026/10/a.pdf'}" in (result.items[0].message or "")
+    assert names(dst) == [] and names(arch) == [] and (src / "a.pdf").exists()
+
+
+def test_without_archive_unchanged(
+    engine: Engine, rule_id: int, dirs: tuple[Path, Path, Path]
+) -> None:
+    src, dst, arch = dirs
+    write(src / "a.pdf")
+    result = run(engine, snap(rule_id), src, dst, None)
+    assert result.items[0].message is None and names(arch) == []
+
+
+# -- Regellauf ----------------------------------------------------------------------------------
+
+
+@pytest.fixture
+def setup(
+    manager: MountManager,
+    fake_backend: FakeMountBackend,
+    make_mount: MountFactory,
+    session_factory: sessionmaker[Session],
+    tmp_path: Path,
+) -> Iterator[tuple[int, Path, Path, Path]]:
+    scans = fake_backend.export("nas", "scans")
+    arch_remote = fake_backend.export("nas", "archiv")
+    src_id = make_mount("scans", share="scans")
+    arch_id = make_mount("archiv", share="archiv")
+    manager.mount(src_id)
+    manager.mount(arch_id)
+    consume = tmp_path / "consume"
+    consume.mkdir()
+    with session_scope(session_factory) as s:
+        r = Rule(
+            name="Scans → Paperless + Archiv",
+            src_mount_id=src_id,
+            dst_path=str(consume),
+            stability_seconds=0,
+            archive_kind=ArchiveKind.MOUNT,
+            archive_mount_id=arch_id,
+            archive_path="scans",
+        )
+        s.add(r)
+        s.flush()
+        rid = r.id
+    yield rid, scans, consume, arch_remote
+
+
+def test_runner_with_archive_mount(
+    setup: tuple[int, Path, Path, Path],
+    manager: MountManager,
+    session_factory: sessionmaker[Session],
+) -> None:
+    rid, scans, consume, arch = setup
+    write(scans / "doc.pdf", b"%PDF")
+    result = RuleRunner(session_factory, manager, sleep=lambda _: None).run(rid)
+    assert result.status is RunStatus.OK, result.message
+    assert (
+        (consume / "doc.pdf").read_bytes() == b"%PDF" == (arch / "scans" / "doc.pdf").read_bytes()
+    )
+    assert not (scans / "doc.pdf").exists()
+
+
+def test_unavailable_archive_makes_rule_wait(
+    setup: tuple[int, Path, Path, Path],
+    manager: MountManager,
+    fake_backend: FakeMountBackend,
+    session_factory: sessionmaker[Session],
+) -> None:
+    rid, scans, consume, _arch = setup
+    write(scans / "doc.pdf")
+    fake_backend.vanish(manager.mountpoint_for("archiv"))
+    result = RuleRunner(session_factory, manager, sleep=lambda _: None).run(rid)
+    assert result.unavailable and result.status is RunStatus.ERROR
+    assert (scans / "doc.pdf").exists() and list(consume.iterdir()) == []
+
+
+def test_mount_used_as_archive_cannot_be_deleted(
+    setup: tuple[int, Path, Path, Path],
+    session_factory: sessionmaker[Session],
+    manager: MountManager,
+) -> None:
+    from netmover.admin import AdminError, delete_mount
+    from netmover.models import Mount
+
+    with session_scope(session_factory) as s:
+        arch_id = s.scalar(select(Mount.id).where(Mount.name == "archiv"))
+        assert arch_id is not None
+        with pytest.raises(AdminError, match="Wird noch von Regeln benutzt"):
+            delete_mount(s, manager, arch_id)
+
+
+def test_rule_form_archive_validation(
+    session_factory: sessionmaker[Session], make_mount: MountFactory
+) -> None:
+    from netmover.admin import parse_rule_form, save_rule
+
+    mid = make_mount()
+    base = {
+        "name": "x",
+        "src_mount_id": str(mid),
+        "dst_kind": "local",
+        "dst_path": "/srv/consume",
+        "interval_seconds": "30",
+        "stability_seconds": "5",
+        "op_timeout_seconds": "120",
+    }
+    with session_scope(session_factory) as s:
+        _, err = parse_rule_form(
+            s, {**base, "archive_kind": "local", "archive_path": "/srv/consume"}
+        )
+        assert "derselbe Ordner" in err["archive_path"]
+        _, err = parse_rule_form(s, {**base, "archive_kind": "mount", "archive_path": "a"})
+        assert "archive_mount_id" in err
+        _, err = parse_rule_form(s, {**base, "archive_kind": "local", "archive_path": "relativ"})
+        assert "absoluten Pfad" in err["archive_path"]
+        d, err = parse_rule_form(
+            s,
+            {
+                **base,
+                "archive_kind": "mount",
+                "archive_mount_id": str(mid),
+                "archive_path": "archiv/scans",
+                "archive_layout": "date",
+            },
+        )
+        assert err == {}
+        r, _ = save_rule(s, d)
+        assert r is not None and r.archive_kind is ArchiveKind.MOUNT
+        assert r.archive_path == "archiv/scans" and r.archive_layout.value == "date"