US06-02: Transfer, Verify, and Remove Active Sources (#78)

photo_pipeline/services/archive_journal.py

@@ -0,0 +1,337 @@
+"""Archive transfer journal — the durable record of every per-file transition
+(US06-02).
+
+Archiving is the only stage that deletes an original, so the journal exists to make
+one question answerable after any crash: *may this source file be removed?* Intent
+is written before the mutation it describes, and the recorded state plus the real
+files on disk are the sole basis for answering it later. This module owns the state
+machine, the durable writes, and the evidence table; it never touches a photo
+(:mod:`photo_pipeline.services.archive_transfer` does).
+
+Per-item state machine (concept §9 "Transfer and removal semantics"):
+
+```
+planned → transferring → verified → removing → complete
+              ↘             ↘          ↘ failed
+```
+
+- ``transferring`` — intent recorded; a temporary copy may exist, the destination
+  may or may not have been published. Nothing has been removed.
+- ``verified``     — the archived bytes exist at their final path, hash exactly as
+  recorded, and the manifest entry is durable. Only from here may a source go.
+- ``removing``     — the source removal is committed to; the source may already be
+  gone while the database still points at it.
+- ``complete``     — source absent, database updated, availability recorded.
+
+``classify`` labels each incomplete item from the journal plus disk evidence:
+
+- ``resumable`` — nothing was published; the source is intact, so applying again is
+  safe.
+- ``forward``   — the archived copy exists and matches its recorded hash, so the
+  remaining steps (manifest, removal, bookkeeping) can be finished deterministically.
+- ``manual``    — the evidence contradicts the journal (missing archive copy, wrong
+  bytes, source and archive both gone). Nothing is guessed and nothing is removed;
+  the item blocks unrelated mutations until a human decides.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from pathlib import Path
+
+from sqlalchemy import select
+from sqlalchemy.orm import sessionmaker
+
+from photo_pipeline.models import ArchiveOperation, ArchivePlan
+from photo_pipeline.services.hashing import sha256_file
+
+
+class ArchiveState:
+    PLANNED = "planned"
+    TRANSFERRING = "transferring"
+    VERIFIED = "verified"
+    REMOVING = "removing"
+    COMPLETE = "complete"
+    FAILED = "failed"
+
+
+ALLOWED_TRANSITIONS = {
+    ArchiveState.PLANNED: {ArchiveState.TRANSFERRING, ArchiveState.FAILED},
+    # From `transferring` the outcome is unknown until evidence is gathered, so it
+    # may resolve forward, back to planned (proven nothing was published), or fail.
+    ArchiveState.TRANSFERRING: {
+        ArchiveState.VERIFIED,
+        ArchiveState.PLANNED,
+        ArchiveState.FAILED,
+    },
+    ArchiveState.VERIFIED: {ArchiveState.REMOVING, ArchiveState.FAILED},
+    # No path back: once the source may be gone, only finishing is safe.
+    ArchiveState.REMOVING: {ArchiveState.COMPLETE, ArchiveState.FAILED},
+    ArchiveState.COMPLETE: set(),
+    # A retry re-enters `transferring`, which rechecks every precondition from
+    # scratch; recovery may also reset a failed item to `planned`.
+    ArchiveState.FAILED: {ArchiveState.PLANNED, ArchiveState.TRANSFERRING},
+}
+
+TERMINAL_STATES = frozenset({ArchiveState.COMPLETE})
+# States where this item may already have touched the filesystem.
+UNSAFE_STATES = frozenset({ArchiveState.TRANSFERRING, ArchiveState.VERIFIED, ArchiveState.REMOVING})
+
+RESUMABLE = "resumable"
+FORWARD = "forward"
+MANUAL = "manual"
+
+
+class JournalError(RuntimeError):
+    pass
+
+
+class InvalidTransition(JournalError):
+    pass
+
+
+class JournalConflict(JournalError):
+    """Fencing check failed; a newer owner has taken over this operation."""
+
+
+def can_transition(current: str, target: str) -> bool:
+    return target in ALLOWED_TRANSITIONS.get(current, set())
+
+
+def _now() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+class ArchiveJournal:
+    def __init__(self, session_factory: sessionmaker) -> None:
+        self._session_factory = session_factory
+
+    # ── intent ────────────────────────────────────────────────────────────────
+
+    def begin(self, operation_id: str, *, worker_id: str, fencing_token: int) -> dict:
+        """Record the intent to transfer **before** touching the filesystem."""
+        with self._session_factory() as session:
+            row = self._require(session, operation_id)
+            if row.fencing_token is not None and fencing_token < row.fencing_token:
+                raise JournalConflict(
+                    f"stale fencing token {fencing_token} (current {row.fencing_token})"
+                )
+            if row.journal_state in TERMINAL_STATES:
+                raise InvalidTransition(f"{row.journal_state} is terminal")
+            if row.journal_state != ArchiveState.TRANSFERRING and not can_transition(
+                row.journal_state, ArchiveState.TRANSFERRING
+            ):
+                raise InvalidTransition(f"{row.journal_state} -> {ArchiveState.TRANSFERRING}")
+            if row.journal_state != ArchiveState.TRANSFERRING:
+                row.attempt_count += 1
+            row.journal_state = ArchiveState.TRANSFERRING
+            row.worker_id = worker_id
+            row.fencing_token = fencing_token
+            row.error_code = row.error_message = None
+            row.updated_at = _now()
+            session.commit()
+            return _operation_dict(row)
+
+    # ── transitions ───────────────────────────────────────────────────────────
+
+    def transition(
+        self,
+        operation_id: str,
+        target: str,
+        *,
+        fencing_token: int | None = None,
+        error: tuple[str, str] | None = None,
+        same_filesystem: bool | None = None,
+    ) -> dict:
+        """Move one operation to ``target``, enforcing the state machine.
+
+        Re-entering the state an operation already holds is a no-op, which is what
+        makes recovery idempotent across repeated restarts.
+        """
+        with self._session_factory() as session:
+            row = self._require(session, operation_id)
+            if fencing_token is not None and row.fencing_token is not None:
+                if fencing_token < row.fencing_token:
+                    raise JournalConflict(
+                        f"stale fencing token {fencing_token} (current {row.fencing_token})"
+                    )
+            if same_filesystem is not None:
+                row.same_filesystem = same_filesystem
+            if row.journal_state == target:
+                session.commit()
+                return _operation_dict(row)  # idempotent
+            if not can_transition(row.journal_state, target):
+                raise InvalidTransition(f"{row.journal_state} -> {target}")
+
+            row.journal_state = target
+            row.updated_at = _now()
+            if target == ArchiveState.VERIFIED:
+                row.verified_at = _now()
+            if target == ArchiveState.COMPLETE:
+                row.removed_at = _now()
+            if error:
+                row.error_code, row.error_message = error[0], error[1][:500]
+            elif target != ArchiveState.FAILED:
+                row.error_code = row.error_message = None
+            session.commit()
+            return _operation_dict(row)
+
+    # ── reads ─────────────────────────────────────────────────────────────────
+
+    def get(self, operation_id: str) -> dict | None:
+        with self._session_factory() as session:
+            row = session.get(ArchiveOperation, operation_id)
+            return _operation_dict(row) if row else None
+
+    def operations(self, plan_id: str) -> list[dict]:
+        with self._session_factory() as session:
+            rows = session.scalars(
+                select(ArchiveOperation)
+                .where(ArchiveOperation.plan_id == plan_id)
+                .order_by(ArchiveOperation.sequence)
+            )
+            return [_operation_dict(row) for row in rows]
+
+    def incomplete(self) -> list[dict]:
+        """Every operation left in a non-terminal, non-planned state — the work a
+        restart has to reason about."""
+        with self._session_factory() as session:
+            rows = session.scalars(
+                select(ArchiveOperation)
+                .where(
+                    ArchiveOperation.journal_state.not_in([*TERMINAL_STATES, ArchiveState.PLANNED])
+                )
+                .order_by(ArchiveOperation.plan_id, ArchiveOperation.sequence)
+            )
+            return [_operation_dict(row) for row in rows]
+
+    # ── startup classification ────────────────────────────────────────────────
+
+    def classify(self, operation_id: str) -> dict:
+        """Classify one incomplete operation from the journal plus disk evidence.
+
+        Hashes the archived copy when one exists: "a file is at the destination" is
+        not evidence that the *right* bytes are, and only the right bytes justify
+        removing an original. Never mutates anything.
+        """
+        row = self.get(operation_id)
+        if row is None:
+            raise JournalError(f"unknown archive operation {operation_id!r}")
+        source = Path(row["source_path"])
+        destination = Path(row["destination_path"])
+        source_exists = source.exists()
+        destination_exists = destination.exists()
+        destination_matches = (
+            destination_exists and sha256_file(destination) == row["expected_sha256"]
+        )
+
+        classification, reason = _classify(
+            row["journal_state"], source_exists, destination_exists, destination_matches
+        )
+        return {
+            "operation_id": operation_id,
+            "plan_id": row["plan_id"],
+            "album": row["album"],
+            "asset_id": row["asset_id"],
+            "source_path": row["source_path"],
+            "destination_path": row["destination_path"],
+            "journal_state": row["journal_state"],
+            "classification": classification,
+            "reason": reason,
+            "source_exists": source_exists,
+            "destination_exists": destination_exists,
+            "destination_matches": destination_matches,
+        }
+
+    def classify_all(self) -> list[dict]:
+        return [self.classify(row["id"]) for row in self.incomplete()]
+
+    def blocks_mutation(self) -> bool:
+        """True when any item may have the library half-archived."""
+        return any(row["journal_state"] in UNSAFE_STATES for row in self.incomplete())
+
+    # ── plan-level ────────────────────────────────────────────────────────────
+
+    def plan_state(self, plan_id: str) -> str:
+        """Derive the plan's state from its items, so the summary can never disagree
+        with the journal."""
+        states = {row["journal_state"] for row in self.operations(plan_id)}
+        if not states:
+            return "planned"
+        if states <= {ArchiveState.COMPLETE}:
+            return "complete"
+        if states & {ArchiveState.FAILED}:
+            return "failed"
+        if states & UNSAFE_STATES:
+            return "applying"
+        return "planned"
+
+    def sync_plan_state(self, plan_id: str) -> str:
+        state = self.plan_state(plan_id)
+        with self._session_factory() as session:
+            plan = session.get(ArchivePlan, plan_id)
+            if plan is None:
+                raise JournalError(f"unknown archive plan {plan_id!r}")
+            if plan.state != state:
+                plan.state = state
+                plan.version += 1
+                plan.updated_at = _now()
+                if state == "complete" and plan.completed_at is None:
+                    plan.completed_at = _now()
+                session.commit()
+        return state
+
+    @staticmethod
+    def _require(session, operation_id: str) -> ArchiveOperation:
+        row = session.get(ArchiveOperation, operation_id)
+        if row is None:
+            raise JournalError(f"unknown archive operation {operation_id!r}")
+        return row
+
+
+def _classify(
+    state: str, source_exists: bool, destination_exists: bool, destination_matches: bool
+) -> tuple[str, str]:
+    """The evidence table. Kept a pure function so every combination is testable."""
+    if destination_exists and not destination_matches and state != ArchiveState.PLANNED:
+        # Someone else's file, or a partial/edited copy: never overwrite it, and
+        # never treat it as the durable archive that justifies a deletion.
+        return MANUAL, "the archived path holds bytes that are not the recorded ones"
+
+    if state in (ArchiveState.TRANSFERRING, ArchiveState.FAILED):
+        if destination_matches:
+            return FORWARD, "the archived copy is durable; finish the remaining steps"
+        if source_exists:
+            return RESUMABLE, "nothing was published; the source is intact"
+        return MANUAL, "neither the source nor a verified archive copy is present"
+
+    if state in (ArchiveState.VERIFIED, ArchiveState.REMOVING):
+        if destination_matches:
+            return FORWARD, "the archived copy is durable; finish the remaining steps"
+        return MANUAL, f"journal says {state} but the archived copy is missing"
+
+    return MANUAL, f"unhandled journal state {state}"
+
+
+def _operation_dict(row: ArchiveOperation) -> dict:
+    return {
+        "id": row.id,
+        "plan_id": row.plan_id,
+        "sequence": row.sequence,
+        "album": row.album,
+        "asset_id": row.asset_id,
+        "source_path": row.source_path,
+        "destination_path": row.destination_path,
+        "archive_path": row.archive_path,
+        "expected_sha256": row.expected_sha256,
+        "byte_size": row.byte_size,
+        "same_filesystem": row.same_filesystem,
+        "journal_state": row.journal_state,
+        "attempt_count": row.attempt_count,
+        "fencing_token": row.fencing_token,
+        "worker_id": row.worker_id,
+        "verified_at": row.verified_at.isoformat() if row.verified_at else None,
+        "removed_at": row.removed_at.isoformat() if row.removed_at else None,
+        "error_code": row.error_code,
+        "error_message": row.error_message,
+    }

photo_pipeline/services/archive_transfer.py

@@ -0,0 +1,605 @@
+"""Transfer an approved archive plan, verify it, and remove the active sources
+(US06-02).
+
+This is the only module that deletes originals from the photo library, so every
+step exists to make one promise keepable: **a source is removed only after the
+archived bytes are durable and proven identical.** The journal (US06-02,
+:mod:`photo_pipeline.services.archive_journal`) records intent before each mutation;
+this module performs the mutations and the recovery that reads that intent back.
+
+Per file the sequence is:
+
+```
+journal.begin (transferring)   ← intent persisted BEFORE any disk change
+recheck preconditions          ← source hash, free destination, no symlink
+copy to a temporary file       ← same directory, so the publish is atomic
+fsync, close, hash it back     ← read from disk; the write is not the evidence
+atomically publish             ← rename onto the final archive path
+append the manifest entry      ← durable on the medium itself, fsynced
+journal → verified
+journal → removing             ← intent to delete, persisted first
+re-verify the archive copy, remove the source, verify its absence
+current_path=NULL, close the path occurrence, availability + location recorded
+journal → complete
+```
+
+Same-filesystem albums may skip the copy and use an atomic ``rename`` instead
+(concept §9), but only when the shared device is proven at run time — never from the
+plan's stored guess — and the published file is hashed afterwards exactly as in the
+copy path.
+
+Rules that are never relaxed:
+
+- An occupied destination is never overwritten; the item fails with the source
+  untouched.
+- A source whose bytes no longer match the plan is never archived and never removed.
+- A crash resolves from journal + disk evidence only: an archive copy that is
+  missing or hashes differently blocks the item for a human instead of being
+  retried or, worse, treated as a successful archive.
+- Recovery is idempotent — repeated passes converge on the same state.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import shutil
+import uuid
+from datetime import datetime, timezone
+from pathlib import Path
+
+from sqlalchemy import select
+from sqlalchemy.orm import sessionmaker
+
+from photo_pipeline.config import Config
+from photo_pipeline.models import ArchiveLocation, ArchiveOperation, ArchivePlan, Asset, AssetPath
+from photo_pipeline.services.archive_journal import (
+    MANUAL,
+    RESUMABLE,
+    ArchiveJournal,
+    ArchiveState,
+)
+from photo_pipeline.services.archives import MARKER_NAME, ArchiveError, ArchiveService
+from photo_pipeline.services.hashing import sha256_file
+from photo_pipeline.services.rename_apply import PreconditionFailed, maybe_fault
+
+# The per-medium manifest: one JSON line per archived file, appended and fsynced
+# before its source is removed. It lives with the bytes so the archive can still be
+# read back if the database is lost.
+MANIFEST_NAME = "archive-manifest.jsonl"
+MANIFEST_VERSION = 1
+TEMP_SUFFIX = ".part"
+TEMP_PREFIX = ".archive-"
+# Fault barrier between "the source is gone" and "the database knows it" — not a
+# journal state, but the transition crash tests care about most.
+SOURCE_REMOVED = "source_removed"
+
+APPLYABLE_PLAN_STATES = frozenset({"planned", "applying", "failed", "complete"})
+
+
+def _now() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+class ArchiveTransferService:
+    def __init__(self, session_factory: sessionmaker, *, config: Config) -> None:
+        self._session_factory = session_factory
+        self._config = config
+        self.journal = ArchiveJournal(session_factory)
+
+    # ── plans ─────────────────────────────────────────────────────────────────
+
+    def create(self, location_id: str, albums: list[str] | None = None, *, token: str) -> dict:
+        """Turn an approved preflight into a durable plan.
+
+        The token is re-derived from a fresh preflight, so a plan can only be
+        created for the exact scope, bytes, and destination the user approved.
+        """
+        preflight = ArchiveService(self._session_factory, config=self._config).preflight(
+            location_id, albums
+        )
+        if not token or token != preflight["token"]:
+            raise ArchiveError("stale_token", "the archive preflight changed since it was approved")
+        if preflight["state"] != "ready":
+            codes = ", ".join(sorted({issue["code"] for issue in preflight["blockers"]})) or "-"
+            raise ArchiveError("blocked", f"the archive scope is blocked: {codes}")
+
+        root = Path(preflight["location"]["root"])
+        plan_id = str(uuid.uuid4())
+        with self._session_factory() as session:
+            session.add(
+                ArchivePlan(
+                    id=plan_id,
+                    location_id=location_id,
+                    token=token,
+                    albums=json.dumps(albums) if albums is not None else None,
+                    state="planned",
+                    schema_version=MANIFEST_VERSION,
+                    asset_count=preflight["totals"]["assets"],
+                    byte_size=preflight["totals"]["bytes"],
+                )
+            )
+            session.flush()  # the plan row must exist before its items reference it
+            sequence = 0
+            for album in preflight["albums"]:
+                destination_dir = Path(album["destination"])
+                for asset in album["assets"]:
+                    source = Path(asset["current_path"])
+                    destination = destination_dir / source.name
+                    session.add(
+                        ArchiveOperation(
+                            id=str(uuid.uuid4()),
+                            plan_id=plan_id,
+                            sequence=sequence,
+                            album=album["album"],
+                            asset_id=asset["asset_id"],
+                            source_path=str(source),
+                            destination_path=str(destination),
+                            archive_path=str(destination.relative_to(root)),
+                            expected_sha256=asset["current_sha256"],
+                            byte_size=asset["byte_size"],
+                            # Recorded as a preview only; the real decision is made
+                            # against the devices at apply time.
+                            same_filesystem=album["transfer_method"] == "move",
+                            journal_state=ArchiveState.PLANNED,
+                        )
+                    )
+                    sequence += 1
+            session.commit()
+        return self.get(plan_id)
+
+    def get(self, plan_id: str) -> dict | None:
+        with self._session_factory() as session:
+            plan = session.get(ArchivePlan, plan_id)
+            if plan is None:
+                return None
+            report = _plan_dict(plan)
+        report["operations"] = self.journal.operations(plan_id)
+        return report
+
+    def list(self) -> list[dict]:
+        with self._session_factory() as session:
+            rows = session.scalars(select(ArchivePlan).order_by(ArchivePlan.created_at))
+            return [_plan_dict(row) for row in rows]
+
+    # ── apply ─────────────────────────────────────────────────────────────────
+
+    def apply(
+        self,
+        plan_id: str,
+        *,
+        expected_version: int | None = None,
+        worker_id: str = "archive",
+    ) -> dict:
+        """Archive every item of a plan, then report what happened.
+
+        Items are independent: one failure records its reason and leaves that
+        source in place; the rest of the album continues.
+        """
+        plan = self._require_plan(plan_id)
+        if expected_version is not None and plan["version"] != expected_version:
+            raise ArchiveError(
+                "stale_plan",
+                f"plan {plan_id} is at version {plan['version']}, expected {expected_version}",
+            )
+        if plan["state"] not in APPLYABLE_PLAN_STATES:
+            raise ArchiveError("invalid_state", f"plan {plan_id} is {plan['state']}")
+
+        # One archiver lane: never start while another plan may be half-archived
+        # (concept §16 lock hierarchy).
+        blocking = [row for row in self.journal.incomplete() if row["plan_id"] != plan_id]
+        if blocking:
+            raise ArchiveError(
+                "archive_pending",
+                f"another archive operation is unresolved ({blocking[0]['id']}); recover it first",
+            )
+
+        token = self._claim_plan(plan_id)
+        location = self._location(plan["location_id"])
+        archived = failed = skipped = 0
+        for operation in self.journal.operations(plan_id):
+            if operation["journal_state"] == ArchiveState.COMPLETE:
+                skipped += 1  # repeated apply is a no-op for finished work
+                continue
+            try:
+                if operation["journal_state"] in (ArchiveState.VERIFIED, ArchiveState.REMOVING):
+                    # The bytes are already archived; never transfer them twice.
+                    self._finish(operation, location, token=token, worker_id=worker_id)
+                else:
+                    self._archive_one(operation, location, token=token, worker_id=worker_id)
+                archived += 1
+            except PreconditionFailed as error:
+                self._fail(operation, token, error.code, str(error))
+                failed += 1
+            except Exception as error:  # unexpected: record and stop touching disk
+                self._fail(operation, token, "archive_error", str(error))
+                failed += 1
+        self._prune_empty_sources(plan_id)
+        state = self.journal.sync_plan_state(plan_id)
+        return {
+            "plan_id": plan_id,
+            "archived": archived,
+            "failed": failed,
+            "skipped": skipped,
+            "state": state,
+        }
+
+    def _prune_empty_sources(self, plan_id: str) -> None:
+        """Drop an album folder once every one of its files is archived.
+
+        ``rmdir`` only: a folder that still holds anything at all — an unarchived
+        file, someone else's file, a subfolder — is left exactly as it is.
+        """
+        folders: dict[Path, set[str]] = {}
+        for operation in self.journal.operations(plan_id):
+            folders.setdefault(Path(operation["source_path"]).parent, set()).add(
+                operation["journal_state"]
+            )
+        for folder, states in folders.items():
+            if states == {ArchiveState.COMPLETE}:
+                try:
+                    folder.rmdir()
+                except OSError:
+                    pass  # not empty, or gone already; either way, leave it alone
+
+    def _archive_one(self, operation: dict, location: dict, *, token: int, worker_id: str) -> None:
+        source = Path(operation["source_path"])
+        destination = Path(operation["destination_path"])
+
+        # 1. Intent first — after this point a crash is recoverable from evidence.
+        self.journal.begin(operation["id"], worker_id=worker_id, fencing_token=token)
+        maybe_fault(ArchiveState.TRANSFERRING)
+
+        # 2. Recheck immediately before mutating; the plan's snapshot is not trusted.
+        self._recheck(operation, source, destination, location)
+        destination.parent.mkdir(parents=True, exist_ok=True)
+
+        # 3. Transfer. Same-filesystem is an optimisation, so it has to be proven
+        #    here rather than believed from the plan.
+        same_filesystem = _same_filesystem(source, destination.parent)
+        if same_filesystem:
+            os.rename(source, destination)
+        else:
+            self._copy_and_publish(operation, source, destination)
+        _fsync_dir(destination.parent)
+
+        # 4. The published file is the archive only once it hashes as recorded.
+        if sha256_file(destination) != operation["expected_sha256"]:
+            raise PreconditionFailed(
+                "archive_mismatch", f"{destination} does not hold the expected bytes"
+            )
+        _append_manifest(destination.parent, _manifest_entry(operation, location))
+        self.journal.transition(
+            operation["id"],
+            ArchiveState.VERIFIED,
+            fencing_token=token,
+            same_filesystem=same_filesystem,
+        )
+        maybe_fault(ArchiveState.VERIFIED)
+
+        # 5. Only now may the active source go.
+        self._finish(self.journal.get(operation["id"]), location, token=token, worker_id=worker_id)
+
+    def _copy_and_publish(self, operation: dict, source: Path, destination: Path) -> None:
+        """Cross-filesystem: copy to a temporary file beside the destination, prove
+        its bytes, then publish it atomically. The source is still untouched."""
+        temp = destination.with_name(f"{TEMP_PREFIX}{uuid.uuid4().hex}{TEMP_SUFFIX}")
+        try:
+            with open(source, "rb") as src, open(temp, "wb") as out:
+                shutil.copyfileobj(src, out, 1024 * 1024)
+                out.flush()
+                os.fsync(out.fileno())
+            if sha256_file(temp) != operation["expected_sha256"]:
+                raise PreconditionFailed("copy_mismatch", f"{source} copied with wrong bytes")
+            if destination.exists():
+                raise PreconditionFailed(
+                    "destination_exists", f"{destination} appeared during the transfer"
+                )
+            # ponytail: rename after an exists() check. The archiver lane is single
+            # and local; use O_EXCL/link-based publish if a second writer ever exists.
+            os.rename(temp, destination)
+        finally:
+            temp.unlink(missing_ok=True)
+
+    def _finish(self, operation: dict, location: dict, *, token: int, worker_id: str) -> None:
+        """Drive an item whose archive copy is durable through removal and
+        bookkeeping. Every step is idempotent, so recovery may replay it."""
+        destination = Path(operation["destination_path"])
+        source = Path(operation["source_path"])
+        state = operation["journal_state"]
+
+        if state == ArchiveState.FAILED:
+            # The archive copy is durable even though the attempt ended badly:
+            # re-enter the transfer state so the remaining steps can run.
+            self.journal.transition(operation["id"], ArchiveState.TRANSFERRING, fencing_token=token)
+            state = ArchiveState.TRANSFERRING
+
+        if state == ArchiveState.TRANSFERRING:
+            if sha256_file(destination) != operation["expected_sha256"]:
+                raise PreconditionFailed(
+                    "archive_mismatch", f"{destination} does not hold the expected bytes"
+                )
+            _append_manifest(destination.parent, _manifest_entry(operation, location))
+            self.journal.transition(operation["id"], ArchiveState.VERIFIED, fencing_token=token)
+            state = ArchiveState.VERIFIED
+
+        if state == ArchiveState.VERIFIED:
+            self.journal.transition(operation["id"], ArchiveState.REMOVING, fencing_token=token)
+            maybe_fault(ArchiveState.REMOVING)
+            state = ArchiveState.REMOVING
+
+        if state == ArchiveState.REMOVING:
+            # Re-verify the archived bytes immediately before deleting the original:
+            # this check is the entire justification for the removal.
+            if not destination.exists() or sha256_file(destination) != operation["expected_sha256"]:
+                raise PreconditionFailed(
+                    "archive_unverified", f"{destination} is not a verified archive copy"
+                )
+            if source.exists():
+                if source.is_symlink():
+                    raise PreconditionFailed("symlink", f"{source} became a symlink")
+                if sha256_file(source) != operation["expected_sha256"]:
+                    raise PreconditionFailed(
+                        "source_changed", f"{source} changed; it is not ours to remove"
+                    )
+                source.unlink()
+            if source.exists():
+                raise PreconditionFailed("removal_failed", f"{source} is still present")
+            # The dangerous window: active storage no longer holds the file while the
+            # database still points at it.
+            maybe_fault(SOURCE_REMOVED)
+            self._record_archived(operation, location, destination)
+            self.journal.transition(operation["id"], ArchiveState.COMPLETE, fencing_token=token)
+            maybe_fault(ArchiveState.COMPLETE)
+
+    def _recheck(self, operation: dict, source: Path, destination: Path, location: dict) -> None:
+        if not source.exists():
+            raise PreconditionFailed("source_missing", f"source {source} disappeared")
+        if source.is_symlink() or destination.is_symlink():
+            raise PreconditionFailed("symlink", "refusing to archive through a symlink")
+        if destination.exists():
+            raise PreconditionFailed("destination_exists", f"destination {destination} is occupied")
+        root = Path(location["root"])
+        if root not in destination.parents:
+            raise PreconditionFailed(
+                "destination_escape", f"{destination} is outside the archive location {root}"
+            )
+        if not root.is_dir() or not (root / MARKER_NAME).exists():
+            raise PreconditionFailed("location_offline", f"{root} is not the archive medium")
+        if sha256_file(source) != operation["expected_sha256"]:
+            raise PreconditionFailed(
+                "source_changed", f"{source} changed since the plan was approved"
+            )
+        with self._session_factory() as session:
+            asset = session.get(Asset, operation["asset_id"])
+            if asset is None or asset.current_path != str(source):
+                raise PreconditionFailed(
+                    "asset_moved", f"asset {operation['asset_id']} is no longer at {source}"
+                )
+
+    # ── database ──────────────────────────────────────────────────────────────
+
+    def _record_archived(self, operation: dict, location: dict, destination: Path) -> None:
+        """The original is gone from active storage: drop ``current_path``, close its
+        occurrence, record where the bytes now live, and set availability."""
+        now = _now()
+        with self._session_factory() as session:
+            asset = session.get(Asset, operation["asset_id"])
+            if asset is None:
+                raise PreconditionFailed(
+                    "asset_missing", f"asset {operation['asset_id']} no longer exists"
+                )
+            if asset.current_path:
+                for row in session.scalars(
+                    select(AssetPath).where(
+                        AssetPath.asset_id == asset.id,
+                        AssetPath.path == asset.current_path,
+                        AssetPath.valid_until.is_(None),
+                    )
+                ):
+                    row.valid_until = now
+            recorded = session.scalar(
+                select(AssetPath).where(
+                    AssetPath.asset_id == asset.id, AssetPath.path == str(destination)
+                )
+            )
+            if recorded is None:  # idempotent: recovery may replay this
+                session.add(
+                    AssetPath(
+                        asset_id=asset.id,
+                        path=str(destination),
+                        valid_from=now,
+                        reason="archive",
+                    )
+                )
+            asset.current_path = None
+            asset.availability_state = (
+                "archived_online" if destination.exists() else "archived_offline"
+            )
+            asset.archive_location_id = location["id"]
+            asset.archive_path = operation["archive_path"]
+            asset.state_version += 1
+            asset.updated_at = now
+            session.commit()
+
+    # ── recovery ──────────────────────────────────────────────────────────────
+
+    def recover(self, *, worker_id: str = "archive-recovery") -> dict:
+        """Resolve every incomplete item from journal + disk evidence.
+
+        Idempotent: running it repeatedly converges. Ambiguous (``manual``) work is
+        left exactly as found and keeps blocking unrelated mutations.
+        """
+        results = {"resumed": 0, "completed": 0, "manual": 0}
+        touched: set[str] = set()
+        for verdict in self.journal.classify_all():
+            operation = self.journal.get(verdict["operation_id"])
+            touched.add(operation["plan_id"])
+            token = (operation["fencing_token"] or 0) + 1
+            if verdict["classification"] == MANUAL:
+                results["manual"] += 1
+                continue
+            if verdict["classification"] == RESUMABLE:
+                # Nothing was published: discard the debris and let a later apply
+                # retry the item cleanly.
+                _clean_temp_files(Path(operation["destination_path"]).parent)
+                self.journal.transition(operation["id"], ArchiveState.PLANNED, fencing_token=token)
+                results["resumed"] += 1
+                continue
+            location = self._location(self._require_plan(operation["plan_id"])["location_id"])
+            try:
+                self._finish(operation, location, token=token, worker_id=worker_id)
+                results["completed"] += 1
+            except PreconditionFailed as error:
+                self._fail(operation, token, error.code, str(error))
+                results["manual"] += 1
+        for plan_id in touched:
+            self._prune_empty_sources(plan_id)
+            self.journal.sync_plan_state(plan_id)
+        return results
+
+    def recovery_status(self) -> dict:
+        verdicts = self.journal.classify_all()
+        return {
+            "operations": verdicts,
+            "manual": [v for v in verdicts if v["classification"] == MANUAL],
+            "blocks_mutation": self.journal.blocks_mutation(),
+        }
+
+    # ── helpers ───────────────────────────────────────────────────────────────
+
+    def _fail(self, operation: dict, token: int, code: str, message: str) -> None:
+        self.journal.transition(
+            operation["id"], ArchiveState.FAILED, fencing_token=token, error=(code, message)
+        )
+
+    def _require_plan(self, plan_id: str) -> dict:
+        with self._session_factory() as session:
+            plan = session.get(ArchivePlan, plan_id)
+            if plan is None:
+                raise ArchiveError("unknown_plan", f"unknown archive plan {plan_id!r}")
+            return _plan_dict(plan)
+
+    def _location(self, location_id: str) -> dict:
+        with self._session_factory() as session:
+            location = session.get(ArchiveLocation, location_id)
+            if location is None:
+                raise ArchiveError("unknown_location", f"unknown archive location {location_id!r}")
+            return {"id": location.id, "root": location.root, "media_id": location.media_id}
+
+    def _claim_plan(self, plan_id: str) -> int:
+        """Bump the plan version and use it as this attempt's fencing token, so a
+        worker from a superseded attempt cannot commit."""
+        with self._session_factory() as session:
+            plan = session.get(ArchivePlan, plan_id)
+            plan.version += 1
+            plan.state = "applying"
+            plan.updated_at = _now()
+            token = plan.version
+            session.commit()
+        return token
+
+
+# ── module helpers ───────────────────────────────────────────────────────────
+
+
+def _same_filesystem(source: Path, destination_dir: Path) -> bool:
+    """Proven at run time from the actual devices, never from the plan's preview."""
+    try:
+        return source.stat().st_dev == destination_dir.stat().st_dev
+    except OSError:
+        return False
+
+
+def _fsync_dir(path: Path) -> None:
+    """Make the directory entry itself durable, so the published name survives a
+    power loss and not just the file's data."""
+    fd = os.open(path, os.O_RDONLY)
+    try:
+        os.fsync(fd)
+    except OSError:
+        pass  # some filesystems refuse directory fsync; the data is already synced
+    finally:
+        os.close(fd)
+
+
+def _manifest_entry(operation: dict, location: dict) -> dict:
+    return {
+        "schema_version": MANIFEST_VERSION,
+        "plan_id": operation["plan_id"],
+        "asset_id": operation["asset_id"],
+        "album": operation["album"],
+        "archive_path": operation["archive_path"],
+        "source_path": operation["source_path"],
+        "sha256": operation["expected_sha256"],
+        "byte_size": operation["byte_size"],
+        "media_id": location["media_id"],
+        "archived_at": _now().isoformat(),
+    }
+
+
+def _append_manifest(directory: Path, entry: dict) -> None:
+    """Append one durable manifest line, skipping an entry that is already there.
+
+    The manifest is written before the source is removed, so it is the medium's own
+    record of what it holds even if the database is lost.
+    """
+    path = directory / MANIFEST_NAME
+    # ponytail: rereads the album manifest per file (O(n²) lines for one album).
+    # Keep an in-memory index per plan if an album ever holds enough files to matter.
+    if path.exists():
+        for line in path.read_text(encoding="utf-8").splitlines():
+            try:
+                existing = json.loads(line)
+            except ValueError:
+                continue
+            if (existing.get("asset_id"), existing.get("sha256")) == (
+                entry["asset_id"],
+                entry["sha256"],
+            ):
+                return
+    with open(path, "a", encoding="utf-8") as handle:
+        handle.write(json.dumps(entry, sort_keys=True) + "\n")
+        handle.flush()
+        os.fsync(handle.fileno())
+    _fsync_dir(directory)
+
+
+def read_manifest(directory: Path) -> list[dict]:
+    """Every manifest entry an archived album directory holds."""
+    path = directory / MANIFEST_NAME
+    if not path.exists():
+        return []
+    entries = []
+    for line in path.read_text(encoding="utf-8").splitlines():
+        try:
+            entries.append(json.loads(line))
+        except ValueError:
+            continue
+    return entries
+
+
+def _clean_temp_files(directory: Path) -> None:
+    """Remove this application's own abandoned transfer temporaries — never any
+    other file (concept §17: startup cleans only recognised stale temporaries)."""
+    if not directory.is_dir():
+        return
+    for temp in directory.glob(f"{TEMP_PREFIX}*{TEMP_SUFFIX}"):
+        temp.unlink(missing_ok=True)
+
+
+def _plan_dict(plan: ArchivePlan) -> dict:
+    return {
+        "id": plan.id,
+        "location_id": plan.location_id,
+        "token": plan.token,
+        "albums": json.loads(plan.albums) if plan.albums else None,
+        "state": plan.state,
+        "schema_version": plan.schema_version,
+        "asset_count": plan.asset_count,
+        "byte_size": plan.byte_size,
+        "version": plan.version,
+        "worker_id": plan.worker_id,
+        "completed_at": plan.completed_at.isoformat() if plan.completed_at else None,
+        "created_at": plan.created_at.isoformat() if plan.created_at else None,
+    }

photo_pipeline/services/archives.py

@@ -24,6 +24,7 @@ Preflight proves, per concept §9 "Archive preflight":
   probed by writing them, not assumed.
 
 Blocker codes: ``no_library_root``, ``location_offline``, ``wrong_volume``,
+``archive_pending``,
 ``unsafe_destination``, ``destination_not_writable``, ``manifest_unwritable``,
 ``insufficient_capacity``, ``backup_unavailable``, ``lock_conflict``,
 ``rename_pending``, ``empty_scope``, ``destination_collision``,
@@ -57,6 +58,7 @@ from photo_pipeline.jobs.domain_handlers import ARCHIVE_LOCK, LIBRARY_WRITE_LOCK
 from photo_pipeline.models import ArchiveLocation, Asset, UploadBatch, UploadItem
 from photo_pipeline.path_policy import PathPolicyError, is_excluded, normalize_root, resolve_within
 from photo_pipeline.services.albums import album_label
+from photo_pipeline.services.archive_journal import ArchiveJournal
 from photo_pipeline.services.hashing import sha256_file
 from photo_pipeline.services.jobs import JobService
 from photo_pipeline.services.rename_journal import RenameJournal
@@ -291,6 +293,13 @@ class ArchiveService:
             blockers.append(
                 _issue("rename_pending", "an unresolved rename must be recovered before archiving")
             )
+        if ArchiveJournal(self._session_factory).blocks_mutation():
+            blockers.append(
+                _issue(
+                    "archive_pending",
+                    "an unresolved archive transfer must be recovered before archiving again",
+                )
+            )
         return blockers
 
     def _capacity(self, required: int, probe: dict) -> dict:

photo_pipeline/services/rename_apply.py

@@ -83,12 +83,13 @@ def _now() -> datetime:
     return datetime.now(timezone.utc)
 
 
-def _maybe_fault(state: str) -> None:
+def maybe_fault(state: str) -> None:
     """Test-only crash barrier (concept §18 fault injection).
 
     When ``PHOTO_PIPELINE_FAULT_AFTER`` names a journal state, the process dies
     abruptly the moment that state has been persisted — modelling a real kill at
-    exactly that transition. Never set outside tests.
+    exactly that transition. Never set outside tests. Shared with the archive
+    transfer journal (US06-02), which uses the same env var and its own state names.
     """
     if os.environ.get("PHOTO_PIPELINE_FAULT_AFTER") == state:
         os._exit(9)
@@ -172,7 +173,7 @@ class RenameApplyService:
 
         # 1. Intent first — after this point a crash is recoverable from evidence.
         self.journal.begin(operation["id"], worker_id=worker_id, fencing_token=token)
-        _maybe_fault(JournalState.MOVING)
+        maybe_fault(JournalState.MOVING)
 
         # 2. Recheck preconditions immediately before mutating, never trusting the
         #    plan's snapshot: files can change between preview and confirmation.
@@ -187,21 +188,21 @@ class RenameApplyService:
             os.rename(source, destination)
 
         self.journal.transition(operation["id"], JournalState.MOVED, fencing_token=token)
-        _maybe_fault(JournalState.MOVED)
+        maybe_fault(JournalState.MOVED)
 
         # 4. Database: stable IDs keep their identity, paths are re-pointed and the
         #    old occurrence is closed — all in one transaction.
         self._reconcile_paths(operation, source, destination)
         self.journal.transition(operation["id"], JournalState.DATABASE_UPDATED, fencing_token=token)
-        _maybe_fault(JournalState.DATABASE_UPDATED)
+        maybe_fault(JournalState.DATABASE_UPDATED)
 
         # 5. Postconditions: the bytes really are at the new paths.
         self._verify(operation, destination)
         self.journal.transition(operation["id"], JournalState.VERIFIED, fencing_token=token)
-        _maybe_fault(JournalState.VERIFIED)
+        maybe_fault(JournalState.VERIFIED)
 
         self.journal.transition(operation["id"], JournalState.COMPLETE, fencing_token=token)
-        _maybe_fault(JournalState.COMPLETE)
+        maybe_fault(JournalState.COMPLETE)
 
     def _recheck(self, operation: dict, source: Path, destination: Path) -> None:
         if not source.exists():