US06-04: Plan and Execute Safe Restores

migrations/versions/0011_archive_locations.py

@@ -0,0 +1,40 @@
+"""Archive locations (US06-01).
+
+Revision ID: 0011_archive_locations
+Revises: 0010_upload_verification
+Create Date: 2026-08-16
+
+Configured archive destinations with their stable media identity, last probed
+capabilities, and state.
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = "0011_archive_locations"
+down_revision = "0010_upload_verification"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "archive_locations",
+        sa.Column("id", sa.String(), primary_key=True),
+        sa.Column("name", sa.String(), nullable=False, unique=True),
+        sa.Column("root", sa.String(), nullable=False),
+        sa.Column("media_id", sa.String(), nullable=False, unique=True),
+        sa.Column("capabilities", sa.String(), nullable=True),
+        sa.Column("state", sa.String(), nullable=False, server_default="offline"),
+        sa.Column("last_seen_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column(
+            "created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()
+        ),
+        sa.Column(
+            "updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()
+        ),
+    )
+
+
+def downgrade() -> None:
+    op.drop_table("archive_locations")

migrations/versions/0012_archive_plans.py

@@ -0,0 +1,96 @@
+"""Archive plans, per-item transfer journal, and archived asset location (US06-02).
+
+Revision ID: 0012_archive_plans
+Revises: 0011_archive_locations
+Create Date: 2026-08-16
+
+The journal is what makes removing an original recoverable: every item records its
+source, destination, expected hash, and the state it had reached before the process
+died. Assets gain the archive location and relative path so an offline original is
+still explained rather than looking missing.
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = "0012_archive_plans"
+down_revision = "0011_archive_locations"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "archive_plans",
+        sa.Column("id", sa.String(), primary_key=True),
+        sa.Column(
+            "location_id",
+            sa.String(),
+            sa.ForeignKey("archive_locations.id"),
+            nullable=False,
+            index=True,
+        ),
+        # The preflight token this plan was approved against.
+        sa.Column("token", sa.String(), nullable=False),
+        sa.Column("albums", sa.String(), nullable=True),  # JSON array
+        # planned | applying | complete | failed
+        sa.Column("state", sa.String(), nullable=False, server_default="planned"),
+        sa.Column("schema_version", sa.Integer(), nullable=False, server_default="1"),
+        sa.Column("asset_count", sa.Integer(), nullable=False, server_default="0"),
+        sa.Column("byte_size", sa.Integer(), nullable=False, server_default="0"),
+        # Bumped on every claim and used as the fencing token.
+        sa.Column("version", sa.Integer(), nullable=False, server_default="1"),
+        sa.Column("worker_id", sa.String(), nullable=True),
+        sa.Column("completed_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column(
+            "created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()
+        ),
+        sa.Column(
+            "updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()
+        ),
+    )
+    op.create_table(
+        "archive_operations",
+        sa.Column("id", sa.String(), primary_key=True),
+        sa.Column(
+            "plan_id",
+            sa.String(),
+            sa.ForeignKey("archive_plans.id", ondelete="CASCADE"),
+            nullable=False,
+            index=True,
+        ),
+        sa.Column("sequence", sa.Integer(), nullable=False),
+        sa.Column("album", sa.String(), nullable=False),
+        sa.Column("asset_id", sa.String(), sa.ForeignKey("assets.id"), nullable=False, index=True),
+        sa.Column("source_path", sa.String(), nullable=False),
+        sa.Column("destination_path", sa.String(), nullable=False),
+        # Relative to the location root: the medium can be mounted anywhere later.
+        sa.Column("archive_path", sa.String(), nullable=False),
+        sa.Column("expected_sha256", sa.String(), nullable=False),
+        sa.Column("byte_size", sa.Integer(), nullable=True),
+        sa.Column("same_filesystem", sa.Boolean(), nullable=True),
+        # planned | transferring | verified | removing | complete | failed
+        sa.Column("journal_state", sa.String(), nullable=False, server_default="planned"),
+        sa.Column("attempt_count", sa.Integer(), nullable=False, server_default="0"),
+        sa.Column("fencing_token", sa.Integer(), nullable=True),
+        sa.Column("worker_id", sa.String(), nullable=True),
+        sa.Column("verified_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("removed_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("error_code", sa.String(), nullable=True),
+        sa.Column("error_message", sa.String(), nullable=True),
+        sa.Column(
+            "updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()
+        ),
+        sa.UniqueConstraint("plan_id", "sequence", name="uq_archive_operations_plan_sequence"),
+    )
+    # Plain columns: SQLite cannot ALTER a table to add a foreign key, and the
+    # relationship is enforced by the service that writes them.
+    op.add_column("assets", sa.Column("archive_location_id", sa.String(), nullable=True))
+    op.add_column("assets", sa.Column("archive_path", sa.String(), nullable=True))
+
+
+def downgrade() -> None:
+    op.drop_column("assets", "archive_path")
+    op.drop_column("assets", "archive_location_id")
+    op.drop_table("archive_operations")
+    op.drop_table("archive_plans")

migrations/versions/0013_protected_thumbnails.py

@@ -0,0 +1,29 @@
+"""Protected thumbnails (US06-03).
+
+Revision ID: 0013_protected_thumbnails
+Revises: 0012_archive_plans
+Create Date: 2026-08-16
+
+A protected thumbnail is the durable comparison preview of an asset whose
+original has left active storage. It is evidence rather than cache, so the LRU
+quota must not evict it: the archive medium may be offline when it is needed.
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = "0013_protected_thumbnails"
+down_revision = "0012_archive_plans"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "thumbnails",
+        sa.Column("protected", sa.Boolean(), nullable=False, server_default=sa.false()),
+    )
+
+
+def downgrade() -> None:
+    op.drop_column("thumbnails", "protected")

migrations/versions/0014_restore_plans.py

@@ -0,0 +1,37 @@
+"""Restore plans and archive divergence (US06-04).
+
+Revision ID: 0014_restore_plans
+Revises: 0013_protected_thumbnails
+Create Date: 2026-08-16
+
+Restore reuses the archive plan and journal tables: the crash-safe question is the
+same one in the opposite direction (copy, verify, publish, register), so the rows
+gain a ``direction`` instead of a parallel pair of tables. ``archive_divergent_at``
+records the moment an archived copy was proven to hold bytes that are not the ones
+the database recorded — a restore must never silently accept a different file.
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = "0014_restore_plans"
+down_revision = "0013_protected_thumbnails"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    for table in ("archive_plans", "archive_operations"):
+        op.add_column(
+            table,
+            sa.Column("direction", sa.String(), nullable=False, server_default="archive"),
+        )
+    op.add_column(
+        "assets", sa.Column("archive_divergent_at", sa.DateTime(timezone=True), nullable=True)
+    )
+
+
+def downgrade() -> None:
+    op.drop_column("assets", "archive_divergent_at")
+    for table in ("archive_plans", "archive_operations"):
+        op.drop_column(table, "direction")

photo_pipeline/api/app.py

@@ -17,6 +17,7 @@ from fastapi.staticfiles import StaticFiles
 from photo_pipeline.api.routes import (
     albums,
     analysis,
+    archives,
     duplicates,
     health,
     inventory,
@@ -73,6 +74,7 @@ def create_app(config: Config | None = None) -> FastAPI:
     app.include_router(albums.router, prefix="/api/v1")
     app.include_router(renames.router, prefix="/api/v1")
     app.include_router(uploads.router, prefix="/api/v1")
+    app.include_router(archives.router, prefix="/api/v1")
     # Static single-page app (hash-routed). Mounted last so /api/v1 wins.
     if FRONTEND_DIR.is_dir():
         app.mount("/app", StaticFiles(directory=FRONTEND_DIR, html=True), name="app")

photo_pipeline/api/routes/archives.py

@@ -0,0 +1,234 @@
+"""Archive location, preflight, plan, and restore API (US06-01, US06-02, US06-04).
+
+Registering a location writes a marker onto the medium; preflight is a command
+rather than a read, because it probes the destination, hashes the scope, and issues
+the token an archive plan must present. Creating a plan writes only database rows —
+the transfer itself runs on the durable ``archive`` lane, never in the request
+thread, because it removes originals.
+"""
+
+from __future__ import annotations
+
+from fastapi import APIRouter, Request
+from fastapi.responses import JSONResponse
+from pydantic import BaseModel
+
+from photo_pipeline.jobs.domain_handlers import ARCHIVE_LOCK, ARCHIVE_PLAN, RESTORE_PLAN
+from photo_pipeline.services.archives import ArchiveError, ArchiveService
+from photo_pipeline.services.archive_transfer import ArchiveTransferService
+from photo_pipeline.services.jobs import JobBlocked, JobService
+from photo_pipeline.services.restores import RestoreService
+
+router = APIRouter(tags=["archives"])
+
+# Which failures are the caller's request (422), a missing thing (404), or state
+# that changed under the caller (409).
+NOT_FOUND_CODES = {"unknown_location", "unknown_plan"}
+CONFLICT_CODES = {"stale_token", "stale_plan", "archive_pending"}
+
+
+class RegisterLocationRequest(BaseModel):
+    name: str
+    root: str
+
+
+class PreflightRequest(BaseModel):
+    location_id: str
+    # ``None`` means every album; an explicit list scopes the check.
+    albums: list[str] | None = None
+
+
+class CreatePlanRequest(PreflightRequest):
+    # The token of the preflight the user approved; a stale one is refused.
+    token: str
+
+
+class RestoreRequest(BaseModel):
+    location_id: str
+    # ``None`` means every asset archived at this location.
+    asset_ids: list[str] | None = None
+
+
+class CreateRestoreRequest(RestoreRequest):
+    token: str
+
+
+def _service(request: Request) -> ArchiveService:
+    return ArchiveService(request.app.state.session_factory, config=request.app.state.config)
+
+
+def _restores(request: Request) -> RestoreService:
+    return RestoreService(request.app.state.session_factory, config=request.app.state.config)
+
+
+def _transfers(request: Request) -> ArchiveTransferService:
+    return ArchiveTransferService(
+        request.app.state.session_factory, config=request.app.state.config
+    )
+
+
+def _error(error: ArchiveError) -> JSONResponse:
+    if error.code in NOT_FOUND_CODES:
+        status = 404
+    elif error.code in CONFLICT_CODES:
+        status = 409
+    else:
+        status = 422
+    return JSONResponse(
+        status_code=status, content={"error": {"code": error.code, "message": str(error)}}
+    )
+
+
+@router.post("/archive-locations", status_code=201)
+def register_location(body: RegisterLocationRequest, request: Request):
+    try:
+        return _service(request).register(body.name, body.root)
+    except ArchiveError as error:
+        return _error(error)
+
+
+@router.get("/archive-locations")
+def list_locations(request: Request) -> dict:
+    return {"locations": _service(request).locations()}
+
+
+@router.post("/archive-preflight")
+def preflight(body: PreflightRequest, request: Request):
+    try:
+        return _service(request).preflight(body.location_id, body.albums)
+    except ArchiveError as error:
+        return _error(error)
+
+
+@router.post("/archive-plans", status_code=201)
+def create_plan(body: CreatePlanRequest, request: Request):
+    """Turn an approved preflight into a durable, journaled plan. Nothing moves."""
+    try:
+        return _transfers(request).create(body.location_id, body.albums, token=body.token)
+    except ArchiveError as error:
+        return _error(error)
+
+
+@router.get("/archive-plans")
+def list_plans(request: Request) -> dict:
+    return {"plans": _transfers(request).list()}
+
+
+@router.get("/archive-plans/{plan_id}")
+def get_plan(plan_id: str, request: Request):
+    plan = _transfers(request).get(plan_id)
+    if plan is None:
+        return _error(ArchiveError("unknown_plan", f"unknown archive plan {plan_id}"))
+    return plan
+
+
+@router.post("/archive-plans/{plan_id}/apply")
+def apply_plan(plan_id: str, request: Request):
+    """Queue the transfer on the archiver lane. The worker removes the sources."""
+    service = _transfers(request)
+    plan = service.get(plan_id)
+    if plan is None:
+        return _error(ArchiveError("unknown_plan", f"unknown archive plan {plan_id}"))
+    if service.journal.blocks_mutation():
+        unresolved = [row for row in service.journal.incomplete() if row["plan_id"] != plan_id]
+        if unresolved:
+            return _error(
+                ArchiveError(
+                    "archive_pending",
+                    f"an unresolved archive operation ({unresolved[0]['id']}) must be recovered",
+                )
+            )
+    try:
+        job = JobService(request.app.state.session_factory).enqueue(
+            ARCHIVE_PLAN,
+            lock=ARCHIVE_LOCK,
+            # One queued attempt per plan version: a double-clicked apply reuses it.
+            idempotency_key=f"archive:{plan_id}:{plan['version']}",
+            items=[plan_id],
+        )
+    except JobBlocked as error:
+        return JSONResponse(
+            status_code=409, content={"error": {"code": error.code, "message": str(error)}}
+        )
+    return {"plan_id": plan_id, "job": job}
+
+
+@router.post("/restore-preflight")
+def restore_preflight(body: RestoreRequest, request: Request):
+    """Validate restoring archived assets back into the library. Nothing moves."""
+    try:
+        return _restores(request).preflight(body.location_id, body.asset_ids)
+    except ArchiveError as error:
+        return _error(error)
+
+
+@router.post("/restore-plans", status_code=201)
+def create_restore_plan(body: CreateRestoreRequest, request: Request):
+    try:
+        return _restores(request).create(body.location_id, body.asset_ids, token=body.token)
+    except ArchiveError as error:
+        return _error(error)
+
+
+@router.get("/restore-plans")
+def list_restore_plans(request: Request) -> dict:
+    return {"plans": _restores(request).list()}
+
+
+@router.get("/restore-plans/{plan_id}")
+def get_restore_plan(plan_id: str, request: Request):
+    plan = _restores(request).get(plan_id)
+    if plan is None:
+        return _error(ArchiveError("unknown_plan", f"unknown restore plan {plan_id}"))
+    return plan
+
+
+@router.post("/restore-plans/{plan_id}/apply")
+def apply_restore_plan(plan_id: str, request: Request):
+    """Queue the restore on the archiver lane — the same single lane as archiving,
+    because both move the same originals."""
+    service = _restores(request)
+    plan = service.get(plan_id)
+    if plan is None:
+        return _error(ArchiveError("unknown_plan", f"unknown restore plan {plan_id}"))
+    unresolved = [row for row in service.journal.incomplete() if row["plan_id"] != plan_id]
+    if unresolved:
+        return _error(
+            ArchiveError(
+                "archive_pending",
+                f"an unresolved archive operation ({unresolved[0]['id']}) must be recovered",
+            )
+        )
+    try:
+        job = JobService(request.app.state.session_factory).enqueue(
+            RESTORE_PLAN,
+            lock=ARCHIVE_LOCK,
+            idempotency_key=f"restore:{plan_id}:{plan['version']}",
+            items=[plan_id],
+        )
+    except JobBlocked as error:
+        return JSONResponse(
+            status_code=409, content={"error": {"code": error.code, "message": str(error)}}
+        )
+    return {"plan_id": plan_id, "job": job}
+
+
+@router.get("/restore-recovery")
+def restore_recovery_status(request: Request) -> dict:
+    return _restores(request).recovery_status()
+
+
+@router.post("/restore-recovery/resolve")
+def resolve_restore_recovery(request: Request) -> dict:
+    return _restores(request).recover()
+
+
+@router.get("/archive-recovery")
+def recovery_status(request: Request) -> dict:
+    """What an interrupted transfer left behind, straight from journal + disk."""
+    return _transfers(request).recovery_status()
+
+
+@router.post("/archive-recovery/resolve")
+def resolve_recovery(request: Request) -> dict:
+    return _transfers(request).recover()

photo_pipeline/config.py

@@ -35,6 +35,9 @@ class Config(BaseModel):
     thumbnail_cache_quota_bytes: int = 500_000_000
     thumbnail_max_pixels: int = 100_000_000
 
+    # Free space an archive destination must keep beyond the transfer itself.
+    archive_free_space_reserve_bytes: int = 1_000_000_000
+
     vision_api_key: SecretStr | None = None
     immich_api_key: SecretStr | None = None
     immich_server_url: str = ""

photo_pipeline/jobs/domain_handlers.py

@@ -1,10 +1,12 @@
-"""Domain job handlers: safety scoring, content analysis, uploads (US02-06, US05-02).
+"""Domain job handlers: safety scoring, content analysis, uploads, archive
+transfers, restores (US02-06, US05-02, US06-02, US06-04).
 
-Importing this module registers the ``safety_score``, ``analysis``, and
-``upload_batch`` job types so the generic worker can run them per item. Each handler
-delegates to its service, which owns the real work and the privacy gate. Handlers
-are idempotent: re-scoring or re-analyzing one asset is safe after an interrupted
-attempt, and an upload batch refuses to re-run an attempt whose outcome is unknown.
+Importing this module registers the ``safety_score``, ``analysis``,
+``upload_batch``, ``archive_plan``, and ``restore_plan`` job types so the generic
+worker can run them per item. Each handler delegates to its service, which owns the real work and the
+privacy gate. Handlers are idempotent: re-scoring or re-analyzing one asset is safe
+after an interrupted attempt, an upload batch refuses to re-run an attempt whose
+outcome is unknown, and an archive plan skips items it already completed.
 
 Providers/models are the service defaults here (real NsfwModel / vision provider);
 tests exercise the services directly with injected fakes rather than the worker.
@@ -17,10 +19,14 @@ from photo_pipeline.jobs.handlers import Cancelled, JobContext, register
 SAFETY_SCORE = "safety_score"
 ANALYSIS = "analysis"
 UPLOAD_BATCH = "upload_batch"
+ARCHIVE_PLAN = "archive_plan"
+RESTORE_PLAN = "restore_plan"
 # Both mutate the library's metadata/derived state; one at a time (concept §one job).
 LIBRARY_WRITE_LOCK = "library_write"
 # The uploader lane: one album batch at a time (concept §16).
 UPLOAD_LOCK = "upload"
+# The archiver lane: one archive/restore plan at a time (concept §16).
+ARCHIVE_LOCK = "archive"
 
 
 def _safety_score_item(asset_id: str, ctx: JobContext) -> None:
@@ -51,6 +57,36 @@ def _upload_batch_item(batch_id: str, ctx: JobContext) -> None:
         raise RuntimeError(f"upload batch {batch_id} is {batch['state']}: {batch['error_code']}")
 
 
+def _archive_plan_item(plan_id: str, ctx: JobContext) -> None:
+    """One item = one archive plan. Item-level failures stay in the journal (the
+    source is then still there); only an unusable plan fails the job."""
+    from photo_pipeline.config import Config
+    from photo_pipeline.services.archive_transfer import ArchiveTransferService
+
+    config = ctx.config if ctx.config is not None else Config.from_env()
+    result = ArchiveTransferService(ctx.session_factory, config=config).apply(
+        plan_id, worker_id=ctx.worker_id
+    )
+    if result["failed"]:
+        raise RuntimeError(f"archive plan {plan_id}: {result['failed']} item(s) failed")
+
+
+def _restore_plan_item(plan_id: str, ctx: JobContext) -> None:
+    """One item = one restore plan. A restore removes nothing, so an item failure
+    simply leaves that asset archived (US06-04)."""
+    from photo_pipeline.config import Config
+    from photo_pipeline.services.restores import RestoreService
+
+    config = ctx.config if ctx.config is not None else Config.from_env()
+    result = RestoreService(ctx.session_factory, config=config).apply(
+        plan_id, worker_id=ctx.worker_id
+    )
+    if result["failed"]:
+        raise RuntimeError(f"restore plan {plan_id}: {result['failed']} item(s) failed")
+
+
 register(SAFETY_SCORE, _safety_score_item)
 register(ANALYSIS, _analysis_item)
 register(UPLOAD_BATCH, _upload_batch_item)
+register(ARCHIVE_PLAN, _archive_plan_item)
+register(RESTORE_PLAN, _restore_plan_item)

photo_pipeline/models/__init__.py

@@ -5,6 +5,7 @@ Alembic environment relies on.
 """
 
 from photo_pipeline.models.albums import AlbumProposal
+from photo_pipeline.models.archives import ArchiveLocation, ArchiveOperation, ArchivePlan
 from photo_pipeline.models.assets import Asset, AssetPath
 from photo_pipeline.models.duplicates import (
     DuplicateCluster,
@@ -19,6 +20,9 @@ from photo_pipeline.models.workflow import AnalysisResult, SafetyReview
 
 __all__ = [
     "AlbumProposal",
+    "ArchiveLocation",
+    "ArchiveOperation",
+    "ArchivePlan",
     "Asset",
     "AssetPath",
     "DuplicateCluster",

photo_pipeline/models/archives.py

@@ -0,0 +1,127 @@
+"""Archive location, plan, and transfer-journal persistence (US06-01, US06-02).
+
+An archive location is a *medium*, not a path. External disks get mounted at
+different mountpoints, and a different disk can be mounted at the same one, so a
+recorded root alone can never prove "these bytes went to that volume". Each
+location therefore owns a marker file written onto the medium itself; its
+``media_id`` is the stable identity, and the root is only where it was last seen.
+
+``capabilities`` and ``state`` are the last probe result, kept so the UI can list
+locations without touching a sleeping disk. Preflight always re-probes — a stored
+state is a hint, never evidence.
+
+An ``ArchivePlan`` is one approved preflight turned into durable work, and each
+``ArchiveOperation`` is one file's crash-safe journal row (US06-02). The row records
+what the transfer *intends* to do before it does it — source, destination, expected
+hash — because after a crash that intent plus the files on disk is the only evidence
+available for deciding whether an original may be removed.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime
+
+from sqlalchemy import (
+    Boolean,
+    DateTime,
+    ForeignKey,
+    Integer,
+    String,
+    UniqueConstraint,
+    func,
+)
+from sqlalchemy.orm import Mapped, mapped_column
+
+from photo_pipeline.db import Base
+
+
+class ArchiveLocation(Base):
+    __tablename__ = "archive_locations"
+
+    id: Mapped[str] = mapped_column(String, primary_key=True)
+    name: Mapped[str] = mapped_column(String, nullable=False, unique=True)
+    root: Mapped[str] = mapped_column(String, nullable=False)
+    # Written into the marker file on the medium; proves the right volume is mounted.
+    media_id: Mapped[str] = mapped_column(String, nullable=False, unique=True)
+    capabilities: Mapped[str | None] = mapped_column(String)  # JSON, last probe
+    # online | offline | wrong_volume | unwritable — the last probe's verdict.
+    state: Mapped[str] = mapped_column(String, nullable=False, default="offline")
+    last_seen_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, server_default=func.now()
+    )
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now()
+    )
+
+
+class ArchivePlan(Base):
+    __tablename__ = "archive_plans"
+
+    id: Mapped[str] = mapped_column(String, primary_key=True)
+    location_id: Mapped[str] = mapped_column(
+        ForeignKey("archive_locations.id"), nullable=False, index=True
+    )
+    # The preflight token this plan was approved against; re-verified before apply.
+    token: Mapped[str] = mapped_column(String, nullable=False)
+    albums: Mapped[str | None] = mapped_column(String)  # JSON array
+    # archive | restore — the same journal read in the opposite direction (US06-04).
+    direction: Mapped[str] = mapped_column(String, nullable=False, default="archive")
+
+    # planned | applying | complete | failed
+    state: Mapped[str] = mapped_column(String, nullable=False, default="planned")
+    schema_version: Mapped[int] = mapped_column(Integer, nullable=False, default=1)
+    asset_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
+    byte_size: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
+    # Bumped on every claim and used as the fencing token, so a superseded attempt
+    # cannot commit.
+    version: Mapped[int] = mapped_column(Integer, nullable=False, default=1)
+    worker_id: Mapped[str | None] = mapped_column(String)
+    completed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, server_default=func.now()
+    )
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now()
+    )
+
+
+class ArchiveOperation(Base):
+    __tablename__ = "archive_operations"
+    __table_args__ = (
+        UniqueConstraint("plan_id", "sequence", name="uq_archive_operations_plan_sequence"),
+    )
+
+    id: Mapped[str] = mapped_column(String, primary_key=True)
+    plan_id: Mapped[str] = mapped_column(
+        ForeignKey("archive_plans.id", ondelete="CASCADE"), nullable=False, index=True
+    )
+    sequence: Mapped[int] = mapped_column(Integer, nullable=False)
+    album: Mapped[str] = mapped_column(String, nullable=False)
+    asset_id: Mapped[str] = mapped_column(ForeignKey("assets.id"), nullable=False, index=True)
+
+    # archive: library → medium. restore: medium → library (US06-04). ``source_path``
+    # and ``destination_path`` always mean "from" and "to" for this direction.
+    direction: Mapped[str] = mapped_column(String, nullable=False, default="archive")
+    source_path: Mapped[str] = mapped_column(String, nullable=False)
+    destination_path: Mapped[str] = mapped_column(String, nullable=False)
+    # Relative to the location root, because the medium can be mounted elsewhere.
+    archive_path: Mapped[str] = mapped_column(String, nullable=False)
+    expected_sha256: Mapped[str] = mapped_column(String, nullable=False)
+    byte_size: Mapped[int | None] = mapped_column(Integer)
+    same_filesystem: Mapped[bool | None] = mapped_column(Boolean)
+
+    # planned | transferring | verified | removing | complete | failed
+    journal_state: Mapped[str] = mapped_column(String, nullable=False, default="planned")
+    attempt_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
+    fencing_token: Mapped[int | None] = mapped_column(Integer)
+    worker_id: Mapped[str | None] = mapped_column(String)
+    verified_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+    removed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+    error_code: Mapped[str | None] = mapped_column(String)
+    error_message: Mapped[str | None] = mapped_column(String)
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now()
+    )

photo_pipeline/models/assets.py

@@ -35,7 +35,18 @@ class Asset(Base):
 
     discovered_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
     missing_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+    # active | archiving | archived_online | archived_offline | restoring |
+    # missing_unexpected (concept §9). ``current_path`` is NULL once archived; the
+    # original is then explained by the location plus its relative archive path.
     availability_state: Mapped[str] = mapped_column(String, nullable=False, default="active")
+    # Not a declared foreign key: SQLite cannot add one to an existing table, so the
+    # link is written and read by the archive service (US06-02).
+    archive_location_id: Mapped[str | None] = mapped_column(String)
+    archive_path: Mapped[str | None] = mapped_column(String)
+    # Set when the archived copy was proven to hold bytes other than the recorded
+    # ones (US06-04). Restore refuses such an asset instead of accepting a different
+    # file; cleared as soon as a verification matches again.
+    archive_divergent_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
     # Duplicate canonical link: NULL when the asset is itself canonical or undecided.
     canonical_asset_id: Mapped[str | None] = mapped_column(ForeignKey("assets.id"))
 

photo_pipeline/models/thumbnails.py

@@ -11,7 +11,7 @@ from __future__ import annotations
 
 from datetime import datetime
 
-from sqlalchemy import DateTime, ForeignKey, Integer, String, func
+from sqlalchemy import Boolean, DateTime, ForeignKey, Integer, String, func
 from sqlalchemy.orm import Mapped, mapped_column
 
 from photo_pipeline.db import Base
@@ -29,6 +29,9 @@ class Thumbnail(Base):
     width: Mapped[int | None] = mapped_column(Integer)
     height: Mapped[int | None] = mapped_column(Integer)
     format: Mapped[str | None] = mapped_column(String)
+    # Durable comparison evidence for an archived asset: never evicted by the LRU
+    # quota, because the original may be on a medium that is no longer reachable.
+    protected: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
     created_at: Mapped[datetime] = mapped_column(
         DateTime(timezone=True), nullable=False, server_default=func.now()
     )

photo_pipeline/services/archive_journal.py

@@ -0,0 +1,359 @@
+"""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
+```
+
+A restore (US06-04) uses the same rows with ``direction='restore'``: it copies from
+the medium back into the library and removes nothing, so it goes ``verified →
+complete`` directly. ``source_path``/``destination_path`` always mean "from"/"to",
+which is why the evidence table below needs no direction of its own.
+
+- ``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},
+}
+
+# A restore removes nothing, so it has no ``removing`` step: a verified published
+# copy is the whole job (US06-04). Keeping this as a separate table means the
+# archive direction still cannot reach ``complete`` without going through removal.
+RESTORE_TRANSITIONS = {
+    **ALLOWED_TRANSITIONS,
+    ArchiveState.VERIFIED: {ArchiveState.COMPLETE, ArchiveState.FAILED},
+}
+
+TERMINAL_STATES = frozenset({ArchiveState.COMPLETE})
+# States where this item may already have touched the filesystem.
+UNSAFE_STATES = frozenset({ArchiveState.TRANSFERRING, ArchiveState.VERIFIED, ArchiveState.REMOVING})
+
+# Which way the bytes move. Same rows, same evidence table, opposite direction.
+ARCHIVE = "archive"
+RESTORE = "restore"
+
+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, direction: str = ARCHIVE) -> bool:
+    table = RESTORE_TRANSITIONS if direction == RESTORE else ALLOWED_TRANSITIONS
+    return target in table.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, row.direction
+            ):
+                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, row.direction):
+                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, *, direction: str | None = None) -> list[dict]:
+        """Every operation left in a non-terminal, non-planned state — the work a
+        restart has to reason about. Without ``direction`` this spans archives and
+        restores, because either one half-done blocks the other."""
+        with self._session_factory() as session:
+            stmt = select(ArchiveOperation).where(
+                ArchiveOperation.journal_state.not_in([*TERMINAL_STATES, ArchiveState.PLANNED])
+            )
+            if direction is not None:
+                stmt = stmt.where(ArchiveOperation.direction == direction)
+            rows = session.scalars(
+                stmt.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"],
+            "direction": row["direction"],
+            "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, *, direction: str | None = None) -> list[dict]:
+        return [self.classify(row["id"]) for row in self.incomplete(direction=direction)]
+
+    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,
+        "direction": row.direction,
+        "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,644 @@
+"""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 (
+    ARCHIVE,
+    MANUAL,
+    RESUMABLE,
+    ArchiveJournal,
+    ArchiveState,
+)
+from photo_pipeline.services.archives import MARKER_NAME, ArchiveError, ArchiveService
+from photo_pipeline.services.duplicates import DuplicateService
+from photo_pipeline.services.hashing import sha256_file
+from photo_pipeline.services.rename_apply import PreconditionFailed, maybe_fault
+from photo_pipeline.services.thumbnails import ThumbnailService
+
+# 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,
+                    direction=ARCHIVE,
+                    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,
+                            direction=ARCHIVE,
+                            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)
+                .where(ArchivePlan.direction == ARCHIVE)
+                .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:
+            copy_verify_publish(source, destination, operation["expected_sha256"])
+        _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 _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"
+                )
+            self._require_evidence(operation["asset_id"], destination)
+            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}"
+                )
+
+    def _require_evidence(self, asset_id: str, source: Path) -> dict:
+        """Review evidence must be durable before the original goes.
+
+        The perceptual hash keeps the asset in the fuzzy index once its bytes are
+        unreachable, and the protected preview is what duplicate review can still
+        look at. Both are read from the freshly verified archive copy, which holds
+        exactly the bytes being archived. A file that cannot be decoded has neither
+        — recorded, not fatal, since its exact hashes remain — but failing to
+        produce a preview from a decodable original stops the removal (concept §9).
+        """
+        DuplicateService(self._session_factory).ensure_phash(asset_id, source=source)
+        preview = ThumbnailService(self._session_factory, self._config).ensure_protected(
+            asset_id, source=source
+        )
+        if preview["state"] == "unavailable":
+            raise PreconditionFailed(
+                "preview_unavailable",
+                f"no durable comparison preview for asset {asset_id} "
+                f"({preview['error_code']})",
+            )
+        return preview
+
+    # ── 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(direction=ARCHIVE):
+            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(direction=ARCHIVE)
+        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 copy_verify_publish(source: Path, destination: Path, expected_sha256: str) -> None:
+    """Copy to a temporary file beside the destination, prove its bytes, then publish
+    it atomically. The source is never touched, so a failure costs nothing.
+
+    Shared by archiving (library → medium) and restoring (medium → library, US06-04):
+    both need the same promise that a published file is either complete and correct
+    or not there at all.
+    """
+    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) != 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 _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,
+        "direction": plan.direction,
+        "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

@@ -0,0 +1,637 @@
+"""ArchiveService — destinations and archive preflight (US06-01).
+
+Archive is the only stage that *removes* originals from the active library, so
+this service does the opposite of removing anything: it registers destinations and
+proves, before a single byte moves, that an album could be archived safely. The
+transfer itself is US06-02.
+
+An archive location is a medium, not a path (see :class:`ArchiveLocation`). A
+marker file on the medium carries its ``media_id``, so a disk mounted at the
+recorded root but holding a different marker is ``wrong_volume`` rather than
+silently accepted — the classic "the external disk came back at the same
+mountpoint" data-loss path.
+
+Preflight proves, per concept §9 "Archive preflight":
+
+- the album's upload is *verified*, not merely process-successful, and its bytes on
+  disk still hash to exactly what was uploaded;
+- the destination medium is mounted, is the right one, is writable, lies outside
+  every library root and every ``_IGNORE/`` tree, and has room for the scope plus a
+  configured reserve;
+- nothing already occupies the destination;
+- no rename/upload/archive lease is held, and no rename is half-applied;
+- a database backup and the archive manifest can really be written — both are
+  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``,
+``upload_unverified``, ``bytes_changed``, ``file_missing``, ``preview_unavailable``.
+
+Preflight also *creates* the durable comparison preview of every asset in scope
+(US06-03): it is the evidence duplicate review falls back on once the original is
+on a medium that may be offline, so it has to exist before the original leaves.
+
+Like upload preflight, the confirmation token is *derived* from the report rather
+than stored: any change to the scope, the bytes, the destination, or the blockers
+produces a different token, so a stale browser confirmation can never apply. Values
+that drift without meaning anything (free space, backup size, timestamps) are left
+out of the digest.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import os
+import shutil
+import sqlite3
+import uuid
+from contextlib import closing
+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.integrations import immich_go_report as report_parser
+from photo_pipeline.jobs.domain_handlers import ARCHIVE_LOCK, LIBRARY_WRITE_LOCK, UPLOAD_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.availability import MARKER_NAME, read_marker as _read_marker
+from photo_pipeline.services.availability import refresh as refresh_availability
+from photo_pipeline.services.hashing import sha256_file
+from photo_pipeline.services.jobs import JobService
+from photo_pipeline.services.rename_journal import RenameJournal
+from photo_pipeline.services.thumbnails import ThumbnailService
+from photo_pipeline.services.upload_reports import VERIFIED
+
+PREFLIGHT_VERSION = 1
+TOKEN_PREFIX = f"v{PREFLIGHT_VERSION}"
+MANIFEST_NAME = "archive-manifest.json"
+
+# Upload outcomes that prove Immich holds these exact bytes. ``skipped``/``failed``/
+# ``unknown`` never qualify: archiving on them would remove the only copy.
+ARCHIVED_OUTCOMES = frozenset(
+    {report_parser.UPLOADED, report_parser.UPGRADED, report_parser.DUPLICATE}
+)
+
+LOCKS = (LIBRARY_WRITE_LOCK, UPLOAD_LOCK, ARCHIVE_LOCK)
+
+
+class ArchiveError(RuntimeError):
+    """The request cannot be carried out (unknown location/album, unsafe root)."""
+
+    def __init__(self, code: str, message: str) -> None:
+        super().__init__(message)
+        self.code = code
+
+
+def _now() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+def _issue(code: str, message: str) -> dict:
+    return {"code": code, "message": message}
+
+
+class ArchiveService:
+    def __init__(self, session_factory: sessionmaker, *, config: Config) -> None:
+        self._session_factory = session_factory
+        self._config = config
+        self._roots = tuple(normalize_root(root) for root in config.library_roots)
+
+    # ── locations ─────────────────────────────────────────────────────────────
+
+    def register(self, name: str, root: str) -> dict:
+        """Register an archive destination and stamp its medium with a marker.
+
+        The marker is what makes the location identifiable later, so registering is
+        the one archive operation that writes to the destination up front.
+        """
+        name = (name or "").strip()
+        if not name:
+            raise ArchiveError("name_required", "an archive location needs a name")
+        path = Path(root).expanduser()
+        if not path.is_dir():
+            raise ArchiveError("root_missing", f"{path} is not an existing directory")
+        path = normalize_root(path)
+        unsafe = self._unsafe_destination(path)
+        if unsafe:
+            raise ArchiveError("unsafe_destination", unsafe)
+
+        marker = _read_marker(path)
+        with self._session_factory() as session:
+            if session.scalar(select(ArchiveLocation).where(ArchiveLocation.name == name)):
+                raise ArchiveError("duplicate_name", f"an archive location named {name!r} exists")
+            if marker and session.scalar(
+                select(ArchiveLocation).where(ArchiveLocation.media_id == marker.get("media_id"))
+            ):
+                raise ArchiveError(
+                    "already_registered", f"{path} already belongs to another archive location"
+                )
+            media_id = marker.get("media_id") if marker else str(uuid.uuid4())
+            error = _probe_write(
+                path / MARKER_NAME,
+                json.dumps({"media_id": media_id, "name": name}, indent=2).encode("utf-8"),
+                keep=True,
+            )
+            if error:
+                raise ArchiveError("destination_not_writable", error)
+            location = ArchiveLocation(
+                id=str(uuid.uuid4()),
+                name=name,
+                root=str(path),
+                media_id=media_id,
+                state="online",
+                last_seen_at=_now(),
+            )
+            location.capabilities = json.dumps(_capabilities(path))
+            session.add(location)
+            session.commit()
+            return self._location_report(location, probe=_probe_location(location))
+
+    def locations(self) -> list[dict]:
+        """Every configured location with a fresh probe of its medium."""
+        with self._session_factory() as session:
+            rows = list(session.scalars(select(ArchiveLocation).order_by(ArchiveLocation.name)))
+            reports = []
+            for location in rows:
+                probe = _probe_location(location)
+                location.state = probe["state"]
+                if probe["state"] == "online":
+                    location.last_seen_at = _now()
+                    location.capabilities = json.dumps(probe["capabilities"])
+                reports.append(self._location_report(location, probe=probe))
+            session.commit()
+        # A medium that just appeared or vanished changes what is readable, so the
+        # archived assets are re-derived from the same probe (US06-03).
+        refresh_availability(self._session_factory)
+        return reports
+
+    # ── preflight ─────────────────────────────────────────────────────────────
+
+    def preflight(self, location_id: str, albums: list[str] | None = None) -> dict:
+        """Validate an archive scope against a destination and issue its token.
+
+        Read-only with respect to the library: it hashes files, probes the
+        destination with its own temporary files, and writes nothing else.
+        """
+        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}")
+            probe = _probe_location(location)
+            location.state = probe["state"]
+            if probe["state"] == "online":
+                location.last_seen_at = _now()
+                location.capabilities = json.dumps(probe["capabilities"])
+            report = {
+                "schema_version": PREFLIGHT_VERSION,
+                "location": self._location_report(location, probe=probe),
+                "blockers": [],
+            }
+            root = Path(location.root)
+            session.commit()
+
+        report["blockers"] += self._destination_blockers(root, probe)
+        report["blockers"] += self._lock_blockers()
+        report["albums"] = self._albums(albums, root, reachable=probe["state"] == "online")
+        report["totals"] = _totals(report["albums"])
+        report["capacity"] = self._capacity(report["totals"]["bytes"], probe)
+        if not report["capacity"]["sufficient"]:
+            report["blockers"].append(
+                _issue(
+                    "insufficient_capacity",
+                    f"{report['totals']['bytes']} B plus a "
+                    f"{self._config.archive_free_space_reserve_bytes} B reserve do not fit in "
+                    f"{report['capacity']['free_bytes']} B of free space",
+                )
+            )
+        report["backup"] = self._backup_probe()
+        if not report["backup"]["ok"]:
+            report["blockers"].append(
+                _issue(
+                    "backup_unavailable",
+                    f"a database backup could not be written: {report['backup']['detail']}",
+                )
+            )
+        report["manifest"] = self._manifest_probe(
+            root, report["albums"], writable=probe["writable"]
+        )
+        if not report["manifest"]["ok"]:
+            report["blockers"].append(
+                _issue(
+                    "manifest_unwritable",
+                    f"the archive manifest could not be written: {report['manifest']['detail']}",
+                )
+            )
+        if not report["albums"]:
+            report["blockers"].append(
+                _issue("empty_scope", "no canonical, active assets are in the selected scope")
+            )
+        report["state"] = (
+            "ready"
+            if not report["blockers"] and all(a["state"] == "ready" for a in report["albums"])
+            else "blocked"
+        )
+        report["token"] = _token(report)
+        report["generated_at"] = _now().isoformat()
+        return report
+
+    def verify_token(self, token: str, location_id: str, albums: list[str] | None = None) -> bool:
+        """True when ``token`` still describes this scope and this destination.
+
+        Recomputed, never looked up: an edited source file, a swapped medium, or a
+        newly occupied destination invalidates it without anything writing to the
+        database.
+        """
+        return bool(token) and token == self.preflight(location_id, albums)["token"]
+
+    # ── destination ───────────────────────────────────────────────────────────
+
+    def _unsafe_destination(self, root: Path) -> str | None:
+        """Why this root may never hold archived originals, or ``None``."""
+        if is_excluded(root):
+            return f"{root} is inside an excluded (_IGNORE/) tree"
+        for library in self._roots:
+            if root == library or library in root.parents or root in library.parents:
+                return f"{root} overlaps the active library root {library}"
+        return None
+
+    def _destination_blockers(self, root: Path, probe: dict) -> list[dict]:
+        blockers: list[dict] = []
+        if not self._roots:
+            blockers.append(_issue("no_library_root", "no library root is configured"))
+        if probe["state"] == "offline":
+            blockers.append(
+                _issue("location_offline", f"the archive medium is not mounted at {root}")
+            )
+        elif probe["state"] == "wrong_volume":
+            blockers.append(
+                _issue(
+                    "wrong_volume",
+                    f"{root} holds a different archive medium ({probe['detail']})",
+                )
+            )
+        unsafe = self._unsafe_destination(root)
+        if unsafe:
+            blockers.append(_issue("unsafe_destination", unsafe))
+        if probe["state"] == "unwritable":
+            blockers.append(
+                _issue("destination_not_writable", f"{root} is not writable: {probe['detail']}")
+            )
+        return blockers
+
+    def _lock_blockers(self) -> list[dict]:
+        """Archive is blocked by any lease that may still be moving bytes or metadata."""
+        blockers: list[dict] = []
+        jobs = JobService(self._session_factory)
+        for lock in LOCKS:
+            held = jobs.blockers(lock)
+            if held:
+                blockers.append(
+                    _issue("lock_conflict", f"the {lock} lane is busy: job {held[0]['id']}")
+                )
+        if RenameJournal(self._session_factory).blocks_mutation():
+            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:
+        reserve = self._config.archive_free_space_reserve_bytes
+        free = probe["free_bytes"]
+        return {
+            "required_bytes": required,
+            "reserve_bytes": reserve,
+            "free_bytes": free,
+            "sufficient": free is not None and free >= required + reserve,
+        }
+
+    def _backup_probe(self) -> dict:
+        """Write a real online backup of the database, then discard it.
+
+        A backup that is merely assumed to be possible is worth nothing on the day
+        the archive removes the originals, so this actually runs SQLite's backup API.
+        """
+        source = self._config.database_path
+        target = source.parent / f".archive-preflight-backup-{uuid.uuid4()}.db"
+        try:
+            with closing(sqlite3.connect(source)) as src, closing(sqlite3.connect(target)) as dst:
+                src.backup(dst)
+            size = target.stat().st_size
+        except (sqlite3.Error, OSError) as error:
+            return {"ok": False, "bytes": None, "detail": str(error)}
+        finally:
+            target.unlink(missing_ok=True)
+        return {"ok": True, "bytes": size, "detail": None}
+
+    def _manifest_probe(self, root: Path, albums: list[dict], *, writable: bool) -> dict:
+        """Prove the manifest can be created by writing this exact content and
+        removing it again. The real manifest is written by the transfer (US06-02)."""
+        manifest = {
+            "schema_version": PREFLIGHT_VERSION,
+            "albums": [
+                {
+                    "album": album["album"],
+                    "destination": album["destination"],
+                    "files": [
+                        {
+                            "asset_id": asset["asset_id"],
+                            "source": asset["current_path"],
+                            "sha256": asset["current_sha256"],
+                            "byte_size": asset["byte_size"],
+                        }
+                        for asset in album["assets"]
+                    ],
+                }
+                for album in albums
+            ],
+        }
+        payload = json.dumps(manifest, indent=2, sort_keys=True).encode("utf-8")
+        if not writable:
+            return {"ok": False, "bytes": len(payload), "detail": "the destination is unavailable"}
+        error = _probe_write(root / f".{MANIFEST_NAME}.probe-{uuid.uuid4()}", payload)
+        return {"ok": error is None, "bytes": len(payload), "detail": error}
+
+    def _location_report(self, location: ArchiveLocation, *, probe: dict) -> dict:
+        return {
+            "id": location.id,
+            "name": location.name,
+            "root": location.root,
+            "media_id": location.media_id,
+            "state": probe["state"],
+            "writable": probe["writable"],
+            "device_id": probe["device_id"],
+            "detail": probe["detail"],
+            "last_seen_at": location.last_seen_at.isoformat() if location.last_seen_at else None,
+        }
+
+    # ── scope ─────────────────────────────────────────────────────────────────
+
+    def _albums(self, requested: list[str] | None, root: Path, *, reachable: bool) -> list[dict]:
+        by_album = self._scope()
+        if requested is not None:
+            unknown = sorted(set(requested) - set(by_album))
+            if unknown:
+                raise ArchiveError("unknown_album", f"unknown album(s): {', '.join(unknown)}")
+            by_album = {name: by_album[name] for name in sorted(set(requested))}
+        return [
+            self._album(name, rows, root, reachable=reachable)
+            for name, rows in sorted(by_album.items())
+        ]
+
+    def _scope(self) -> dict[str, list[dict]]:
+        """Canonical, active assets grouped by album, each with its upload evidence."""
+        with self._session_factory() as session:
+            assets = list(
+                session.scalars(
+                    select(Asset).where(
+                        Asset.canonical_asset_id.is_(None),
+                        Asset.availability_state == "active",
+                        Asset.current_path.is_not(None),
+                    )
+                )
+            )
+            uploads: dict[str, UploadItem] = {}
+            for item, batch in session.execute(
+                select(UploadItem, UploadBatch)
+                .join(UploadBatch, UploadBatch.id == UploadItem.batch_id)
+                .order_by(UploadBatch.created_at)
+            ):
+                if _proves_upload(item, batch):
+                    uploads[item.asset_id] = item  # the latest verified batch wins
+
+        by_album: dict[str, list[dict]] = {}
+        for asset in assets:
+            by_album.setdefault(album_label(asset.current_path, self._roots), []).append(
+                {
+                    "asset_id": asset.id,
+                    "path": asset.current_path,
+                    "byte_size": asset.byte_size,
+                    "upload": uploads.get(asset.id),
+                }
+            )
+        return by_album
+
+    def _album(self, name: str, rows: list[dict], root: Path, *, reachable: bool) -> dict:
+        folder = Path(rows[0]["path"]).parent
+        items = sorted(
+            (self._with_preview(_item(row)) for row in rows),
+            key=lambda item: item["current_path"],
+        )
+        blocked = [item for item in items if item["blockers"]]
+        blockers: list[dict] = []
+
+        destination = root / name
+        try:
+            resolve_within(root, destination)
+        except PathPolicyError as error:
+            blockers.append(_issue("unsafe_destination", str(error)))
+        if reachable and destination.exists() and any(destination.iterdir()):
+            blockers.append(
+                _issue("destination_collision", f"{destination} already exists and is not empty")
+            )
+        if blocked:
+            blockers.append(
+                _issue(
+                    "partial_scope",
+                    f"{len(blocked)} of {len(items)} asset(s) are not archivable; an album is "
+                    "archived whole or not at all",
+                )
+            )
+        return {
+            "album": name,
+            "folder": str(folder),
+            "destination": str(destination),
+            # Same filesystem means the transfer can be an atomic move; anything else
+            # is copy-verify-remove (concept §9).
+            "transfer_method": _transfer_method(folder, root),
+            "asset_count": len(items),
+            "blocked_count": len(blocked),
+            "reclaimable_bytes": sum(item["byte_size"] or 0 for item in items),
+            "state": "blocked" if blockers else "ready",
+            "blockers": blockers,
+            "assets": items,
+        }
+
+    def _with_preview(self, item: dict) -> dict:
+        """Create the durable comparison preview while the original is still here.
+
+        This is the last moment it can be made: once the file is archived and the
+        medium leaves, only the retained preview can answer "is this new photo the
+        same picture?". An original that cannot be decoded at all has no preview to
+        keep — its hashes and metadata stay the evidence — but a preview that fails
+        for any other reason blocks the archive (concept §9).
+        """
+        preview = self._previews().ensure_protected(item["asset_id"])
+        item["preview"] = preview
+        if preview["state"] == "unavailable" and not item["blockers"]:
+            item["blockers"].append(
+                _issue(
+                    "preview_unavailable",
+                    f"a durable comparison preview of {item['current_path']} could not be "
+                    f"created ({preview['error_code']})",
+                )
+            )
+        return item
+
+    def _previews(self) -> ThumbnailService:
+        return ThumbnailService(self._session_factory, self._config)
+
+
+# ── internals ────────────────────────────────────────────────────────────────
+
+
+def _item(row: dict) -> dict:
+    """One asset's archivability: verified upload plus the bytes on disk right now."""
+    path = Path(row["path"])
+    upload: UploadItem | None = row["upload"]
+    blockers: list[dict] = []
+    current_sha256 = None
+
+    if not path.exists():
+        blockers.append(_issue("file_missing", f"{path} is missing"))
+    else:
+        # ponytail: full re-hash of the scope. Gate on (size, mtime_ns) first if a
+        # large album makes this slow — the hash stays the authority.
+        current_sha256 = sha256_file(path)
+
+    if upload is None:
+        blockers.append(
+            _issue("upload_unverified", "a verified Immich upload of these bytes is required")
+        )
+    elif current_sha256 is not None and upload.sha256 and current_sha256 != upload.sha256:
+        blockers.append(
+            _issue("bytes_changed", f"{path} changed since it was uploaded; re-upload it first")
+        )
+
+    return {
+        "asset_id": row["asset_id"],
+        "current_path": str(path),
+        "byte_size": row["byte_size"],
+        "current_sha256": current_sha256,
+        "uploaded_sha256": upload.sha256 if upload else None,
+        "blockers": blockers,
+    }
+
+
+def _proves_upload(item: UploadItem, batch: UploadBatch) -> bool:
+    """Whether this upload item is evidence that Immich holds these exact bytes."""
+    return (
+        batch.outcome_state == VERIFIED
+        and not batch.stale_bytes
+        and not item.changed_after_upload
+        and item.outcome in ARCHIVED_OUTCOMES
+    )
+
+
+def _transfer_method(folder: Path, root: Path) -> str:
+    try:
+        if folder.stat().st_dev == root.stat().st_dev:
+            return "move"
+    except OSError:
+        pass
+    return "copy_verify_remove"
+
+
+def _probe_write(path: Path, payload: bytes, *, keep: bool = False) -> str | None:
+    """Write ``payload`` to ``path``; return the failure detail or ``None``."""
+    try:
+        path.write_bytes(payload)
+    except OSError as error:
+        return str(error)
+    if not keep:
+        try:
+            path.unlink()
+        except OSError as error:
+            return str(error)
+    return None
+
+
+def _capabilities(root: Path) -> dict:
+    usage = shutil.disk_usage(root)
+    return {
+        "device_id": root.stat().st_dev,
+        "total_bytes": usage.total,
+        "writable": os.access(root, os.W_OK),
+    }
+
+
+def _probe_location(location: ArchiveLocation) -> dict:
+    """Is the right medium mounted, and can it take bytes right now?"""
+    root = Path(location.root)
+    blank = {"device_id": None, "free_bytes": None, "total_bytes": None, "capabilities": {}}
+    if not root.is_dir():
+        return {"state": "offline", "writable": False, "detail": f"{root} is not mounted", **blank}
+    marker = _read_marker(root)
+    if marker is None:
+        return {
+            "state": "offline",
+            "writable": False,
+            "detail": f"no archive marker found at {root}",
+            **blank,
+        }
+    if marker.get("media_id") != location.media_id:
+        return {
+            "state": "wrong_volume",
+            "writable": False,
+            "detail": f"marker media_id {marker.get('media_id')!r}",
+            **blank,
+        }
+    capabilities = _capabilities(root)
+    usage = shutil.disk_usage(root)
+    # os.access lies on some filesystems; a real write is the only proof.
+    detail = _probe_write(root / f".archive-write-probe-{uuid.uuid4()}", b"")
+    return {
+        "state": "online" if detail is None else "unwritable",
+        "writable": detail is None,
+        "detail": detail,
+        "device_id": capabilities["device_id"],
+        "free_bytes": usage.free,
+        "total_bytes": usage.total,
+        "capabilities": capabilities,
+    }
+
+
+def _totals(albums: list[dict]) -> dict:
+    return {
+        "albums": len(albums),
+        "ready_albums": sum(1 for album in albums if album["state"] == "ready"),
+        "assets": sum(album["asset_count"] for album in albums),
+        "blocked": sum(album["blocked_count"] for album in albums),
+        "bytes": sum(album["reclaimable_bytes"] for album in albums),
+    }
+
+
+def _token(report: dict) -> str:
+    """Digest of everything the report asserts about the scope and the destination.
+
+    Values that drift without changing what would happen — free space, backup size,
+    timestamps — are excluded so the same situation always yields the same token.
+    """
+    payload = {key: value for key, value in report.items() if key not in ("generated_at", "token")}
+    payload["location"] = {
+        key: value for key, value in payload["location"].items() if key != "last_seen_at"
+    }
+    payload["capacity"] = {
+        key: value for key, value in payload["capacity"].items() if key != "free_bytes"
+    }
+    payload["backup"] = {key: value for key, value in payload["backup"].items() if key != "bytes"}
+    digest = hashlib.sha256(
+        json.dumps(payload, sort_keys=True, ensure_ascii=False, default=str).encode("utf-8")
+    ).hexdigest()
+    return f"{TOKEN_PREFIX}:{digest}"

photo_pipeline/services/availability.py

@@ -0,0 +1,124 @@
+"""Where an asset's bytes are right now (US06-03).
+
+Archiving removes the original from the active library but never removes the
+asset: its identity, hashes, decisions, and evidence stay. This module is the one
+place that answers "can these bytes be read, and if not, why" so inventory,
+duplicate review, thumbnails, and the archive service all give the same answer.
+
+States (concept §9):
+
+- ``active``             — the original is in the active library;
+- ``archived_online``    — the recorded medium is mounted and holds the file;
+- ``archived_offline``   — archived, but the medium is not available right now;
+- ``missing_unexpected`` — neither an active path nor the recorded archive
+  location explains the absence. This is the state that must never be confused
+  with ``archived_offline``: an unmounted disk is normal, a mounted disk with a
+  hole in it is not.
+
+A medium is identified by its marker file, never by its mountpoint, so a
+different disk mounted at the recorded root is offline rather than accepted.
+"""
+
+from __future__ import annotations
+
+import json
+from collections import Counter
+from datetime import datetime, timezone
+from pathlib import Path
+
+from sqlalchemy import select
+from sqlalchemy.orm import Session, sessionmaker
+
+from photo_pipeline.models import ArchiveLocation, Asset
+
+ACTIVE = "active"
+ARCHIVED_ONLINE = "archived_online"
+ARCHIVED_OFFLINE = "archived_offline"
+MISSING_UNEXPECTED = "missing_unexpected"
+ARCHIVED = (ARCHIVED_ONLINE, ARCHIVED_OFFLINE)
+
+MARKER_NAME = ".photo-pipeline-archive.json"
+
+
+def read_marker(root: Path) -> dict | None:
+    """The medium's identity marker, or ``None`` when it is not readable."""
+    try:
+        return json.loads((root / MARKER_NAME).read_text(encoding="utf-8"))
+    except (OSError, ValueError):
+        return None
+
+
+def location_online(location: ArchiveLocation) -> bool:
+    """True only when the *recorded* medium is mounted at its root."""
+    marker = read_marker(Path(location.root))
+    return bool(marker) and marker.get("media_id") == location.media_id
+
+
+def archive_file(session: Session, asset: Asset) -> Path | None:
+    """The archived file's absolute path, whether or not the medium is mounted."""
+    if not asset.archive_location_id or not asset.archive_path:
+        return None
+    location = session.get(ArchiveLocation, asset.archive_location_id)
+    if location is None:
+        return None
+    return Path(location.root) / asset.archive_path
+
+
+def readable_path(session: Session, asset: Asset) -> Path | None:
+    """A path whose bytes can be read now: the active file, else the archive copy."""
+    if asset.current_path and Path(asset.current_path).exists():
+        return Path(asset.current_path)
+    archived = archive_file(session, asset)
+    if archived is None:
+        return None
+    location = session.get(ArchiveLocation, asset.archive_location_id)
+    if not location_online(location) or not archived.exists():
+        return None
+    return archived
+
+
+def state_of(session: Session, asset: Asset, *, online: dict[str, bool] | None = None) -> str:
+    """The availability this asset's storage actually justifies right now."""
+    if asset.current_path:
+        return ACTIVE if Path(asset.current_path).exists() else MISSING_UNEXPECTED
+    if not asset.archive_location_id:
+        return MISSING_UNEXPECTED if asset.availability_state != ACTIVE else ACTIVE
+    location = session.get(ArchiveLocation, asset.archive_location_id)
+    if location is None:
+        return MISSING_UNEXPECTED
+    reachable = (
+        online[location.id] if online and location.id in online else location_online(location)
+    )
+    if not reachable:
+        return ARCHIVED_OFFLINE
+    archived = archive_file(session, asset)
+    # The medium is mounted and identified: the file is either there, or it is
+    # genuinely gone — that is not "offline", it needs a human.
+    return ARCHIVED_ONLINE if archived and archived.exists() else MISSING_UNEXPECTED
+
+
+def refresh(session_factory: sessionmaker) -> dict[str, int]:
+    """Re-derive availability for every archived asset from the media themselves.
+
+    Only archived assets are probed: whether an *active* file is present is the
+    inventory scan's job and costs one stat per library file. Each medium is
+    probed once, not once per asset.
+    """
+    counts: Counter[str] = Counter()
+    now = datetime.now(timezone.utc)
+    with session_factory() as session:
+        online = {
+            location.id: location_online(location)
+            for location in session.scalars(select(ArchiveLocation))
+        }
+        for asset in session.scalars(
+            select(Asset).where(Asset.archive_location_id.is_not(None))
+        ):
+            state = state_of(session, asset, online=online)
+            counts[state] += 1
+            if state != asset.availability_state:
+                asset.availability_state = state
+                asset.state_version += 1
+                asset.updated_at = now
+        session.commit()
+    return dict(counts)

photo_pipeline/services/duplicates.py

@@ -11,6 +11,12 @@ Detection runs in two categories:
   band (NEAR/SIMILAR). These are review candidates: never decided automatically, and
   negative-linked pairs are suppressed so a rejected pair is not re-suggested.
 
+Archived assets stay in both indexes (US06-03): a new active copy of an archived
+original is recognised through its hashes even while the medium is offline, and
+cluster review falls back to the retained protected preview plus hash evidence.
+An exact/pixel match links straight to the archived canonical; a perceptual match
+is a review candidate that names the medium to mount for a pixel-level decision.
+
 Decisions (``canonical`` / ``not_duplicate`` / ``deferred``) persist with evidence,
 use optimistic version checks, are reversible, and can never form a canonical cycle.
 A new content-identical member of an already-decided cluster inherits the established
@@ -34,12 +40,14 @@ from sqlalchemy import func, select
 from sqlalchemy.orm import sessionmaker
 
 from photo_pipeline.models import (
+    ArchiveLocation,
     Asset,
     DuplicateCluster,
     DuplicateMember,
     DuplicateNegativeLink,
+    Thumbnail,
 )
-from photo_pipeline.services import hashing
+from photo_pipeline.services import availability, hashing
 
 NEAR_MAX = 5
 SIMILAR_MAX = 10
@@ -127,18 +135,18 @@ class DuplicateService:
 
     # ── perceptual hash backfill ───────────────────────────────────────────
     def ensure_phashes(self) -> int:
+        """Hash whatever is readable now — an archived asset keeps the hash it
+        already has, and gains one whenever its medium happens to be mounted."""
         updated = 0
         with self._session_factory() as session:
-            assets = session.execute(
-                select(Asset).where(
-                    Asset.availability_state == "active",
-                    Asset.current_path.isnot(None),
-                )
-            ).scalars()
+            assets = session.execute(select(Asset)).scalars()
             for asset in assets:
                 if asset.phash is not None and asset.phash_version == hashing.PHASH_VERSION:
                     continue
-                value = hashing.safe_phash(asset.current_path)
+                source = availability.readable_path(session, asset)
+                if source is None:
+                    continue
+                value = hashing.safe_phash(str(source))
                 if value is not None:
                     asset.phash = value
                     asset.phash_version = hashing.PHASH_VERSION
@@ -146,20 +154,40 @@ class DuplicateService:
             session.commit()
         return updated
 
+    def ensure_phash(self, asset_id: str, *, source=None) -> str | None:
+        """Backfill one asset's perceptual hash while its bytes are still readable.
+
+        Archiving calls this before the original leaves — passing the archive copy
+        as ``source``, since the database does not point at it yet — because an
+        asset without a pHash silently drops out of the fuzzy index the moment its
+        medium is away.
+        """
+        with self._session_factory() as session:
+            asset = session.get(Asset, asset_id)
+            if asset is None:
+                return None
+            if asset.phash is not None and asset.phash_version == hashing.PHASH_VERSION:
+                return asset.phash
+            source = source or availability.readable_path(session, asset)
+            if source is None:
+                return None
+            value = hashing.safe_phash(str(source))
+            if value is not None:
+                asset.phash = value
+                asset.phash_version = hashing.PHASH_VERSION
+                session.commit()
+            return value
+
     # ── detection ──────────────────────────────────────────────────────────
     def detect(self) -> DetectionReport:
         self.ensure_phashes()
         now = datetime.now(timezone.utc)
         report = DetectionReport()
         with self._session_factory() as session:
-            assets = list(
-                session.execute(
-                    select(Asset).where(
-                        Asset.availability_state == "active",
-                        Asset.current_path.isnot(None),
-                    )
-                ).scalars()
-            )
+            # Every known asset stays in the indexes, archived or not: a copy of an
+            # archived original must be recognised as a duplicate rather than
+            # treated as a new photo (concept §9, invariant 12).
+            assets = list(session.execute(select(Asset)).scalars())
             by_id = {a.id: a for a in assets}
             negatives = {
                 _pair(link.asset_a, link.asset_b)
@@ -431,10 +459,19 @@ class DuplicateService:
 
     @staticmethod
     def _recommend_canonical(ids, by_id) -> str:
-        # ponytail: largest file, path as deterministic tie-break. The concept's
-        # richer policy (resolution, least recompression, metadata richness) lands
-        # with the review UI story.
-        return max(ids, key=lambda i: (by_id[i].byte_size or 0, by_id[i].current_path or ""))
+        # ponytail: largest file, then the archived copy, then path as a
+        # deterministic tie-break. Archived wins ties because it is the reviewed,
+        # uploaded original — a fresh active copy must not demote it to a variant.
+        # The concept's richer policy (resolution, least recompression, metadata
+        # richness) lands with the review UI story.
+        return max(
+            ids,
+            key=lambda i: (
+                by_id[i].byte_size or 0,
+                by_id[i].availability_state in availability.ARCHIVED,
+                by_id[i].current_path or by_id[i].archive_path or "",
+            ),
+        )
 
     def _apply_canonical(self, session, cluster, ids, canonical_id):
         for member in session.execute(
@@ -507,9 +544,15 @@ class DuplicateService:
                         "current_path": asset.current_path if asset else None,
                         "byte_size": asset.byte_size if asset else None,
                         "phash": asset.phash if asset else None,
+                        **self._offline_evidence(session, asset),
                     }
                 )
             members.sort(key=lambda m: m["asset_id"])
+            # A full-resolution comparison of an offline original is impossible; the
+            # UI asks for that named medium instead of guessing (concept §9).
+            mount_required = sorted(
+                {m["archive_location"] for m in members if m["requires_mount"]}
+            )
             return {
                 "id": cluster.id,
                 "method": cluster.method,
@@ -519,9 +562,54 @@ class DuplicateService:
                 "canonical_asset_id": cluster.canonical_asset_id,
                 "version": cluster.version,
                 "requires_confirmation": cluster.method == Method.PERCEPTUAL.value,
+                "mount_required": mount_required,
                 "members": members,
             }
 
+    def _offline_evidence(self, session, asset: Asset | None) -> dict:
+        """What review can still rely on when a member's original is not readable."""
+        if asset is None:
+            return {
+                "availability_state": None,
+                "archive_location": None,
+                "archive_location_id": None,
+                "archive_path": None,
+                "preview": {"state": "missing", "protected": False},
+                "requires_mount": False,
+            }
+        location = (
+            session.get(ArchiveLocation, asset.archive_location_id)
+            if asset.archive_location_id
+            else None
+        )
+        preview = self._preview_evidence(session, asset.id)
+        archived = asset.availability_state in availability.ARCHIVED
+        return {
+            "availability_state": asset.availability_state,
+            "archive_location": location.name if location else None,
+            "archive_location_id": asset.archive_location_id,
+            "archive_path": asset.archive_path,
+            "preview": preview,
+            # Offline archived members can still be compared through their retained
+            # preview and hash evidence; only pixel-level review needs the medium.
+            "requires_mount": archived
+            and asset.availability_state == availability.ARCHIVED_OFFLINE
+            and bool(location),
+        }
+
+    @staticmethod
+    def _preview_evidence(session, asset_id: str) -> dict:
+        rows = list(
+            session.execute(select(Thumbnail).where(Thumbnail.asset_id == asset_id)).scalars()
+        )
+        ready = [r for r in rows if r.state == "ready" and r.path]
+        if ready:
+            best = max(ready, key=lambda r: (bool(r.protected), r.size or 0))
+            return {"state": "ready", "protected": bool(best.protected), "size": best.size}
+        if rows:
+            return {"state": "unsupported", "protected": False, "size": rows[0].size}
+        return {"state": "missing", "protected": False, "size": None}
+
     # ── decisions ────────────────────────────────────────────────────────────
     def decide(
         self,

photo_pipeline/services/inventory.py

@@ -13,8 +13,11 @@ renames. Every discovered or absent path is classified as one occurrence:
 - ``missing``          — a known active asset whose file is gone (kept, flagged).
 
 Missing files are never pruned (that would break identity); the asset is retained
-with ``missing_at`` set. Archived assets are left untouched. Rescanning unchanged
-input makes no durable change.
+with ``missing_at`` set and its availability becomes ``missing_unexpected`` —
+nothing explains where the bytes went. Archived assets are left untouched: their
+absence from the active roots is expected, and each scan re-derives whether their
+medium is reachable (:mod:`photo_pipeline.services.availability`). Rescanning
+unchanged input makes no durable change.
 
 Extracted from photo_analyzer.discover_photos/reconcile_moved/prune_missing
 (see donor_ledger.yaml: pa-discovery, pa-prune-missing).
@@ -30,12 +33,12 @@ from enum import Enum
 from pathlib import Path
 from typing import Iterable
 
-from sqlalchemy import func, select
+from sqlalchemy import func, or_, select
 from sqlalchemy.orm import Session, sessionmaker
 
 from photo_pipeline import path_policy
 from photo_pipeline.models import Asset, AssetPath
-from photo_pipeline.services import hashing
+from photo_pipeline.services import availability, hashing
 
 
 class Occurrence(str, Enum):
@@ -60,6 +63,8 @@ def _asset_dict(asset: Asset) -> dict:
         "id": asset.id,
         "current_path": asset.current_path,
         "availability_state": asset.availability_state,
+        "archive_location_id": asset.archive_location_id,
+        "archive_path": asset.archive_path,
         "byte_size": asset.byte_size,
         "current_sha256": asset.current_sha256,
         "pixel_sha256": asset.pixel_sha256,
@@ -101,7 +106,9 @@ class InventoryService:
                 result.asset_ids[str(path)] = asset.id
 
             for asset in assets:
-                if asset.availability_state != "active" or asset.id in seen_ids:
+                # Archived assets are explained by their location, not by the active
+                # roots: a scan must never prune or flag them (concept §9).
+                if asset.availability_state in availability.ARCHIVED or asset.id in seen_ids:
                     continue
                 if asset.current_path and asset.current_path not in discovered_paths:
                     if not Path(asset.current_path).exists():
@@ -109,10 +116,18 @@ class InventoryService:
                             asset.missing_at = now
                             asset.state_version += 1
                             asset.updated_at = now
+                        # Nothing explains this absence — it is not an offline medium.
+                        if asset.availability_state != availability.MISSING_UNEXPECTED:
+                            asset.availability_state = availability.MISSING_UNEXPECTED
+                            asset.state_version += 1
+                            asset.updated_at = now
                         result.occurrences[asset.current_path] = Occurrence.MISSING.value
 
             session.commit()
 
+        # Media may have been mounted or removed since the last scan.
+        availability.refresh(self._session_factory)
+
         result.counts = dict(Counter(result.occurrences.values()))
         return result
 
@@ -132,7 +147,11 @@ class InventoryService:
             if availability:
                 stmt = stmt.where(Asset.availability_state == availability)
             if query:
-                stmt = stmt.where(Asset.current_path.like(f"%{query}%"))
+                like = f"%{query}%"
+                # An archived asset has no active path; it is searched where it lives.
+                stmt = stmt.where(
+                    or_(Asset.current_path.like(like), Asset.archive_path.like(like))
+                )
             total = session.scalar(select(func.count()).select_from(stmt.subquery()))
             rows = session.execute(
                 stmt.order_by(Asset.current_path).limit(limit).offset(offset)
@@ -169,6 +188,7 @@ class InventoryService:
                 self._open_path(session, existing.id, path_str, now, occ.value)
             if existing.missing_at is not None:
                 existing.missing_at = None
+                existing.availability_state = availability.ACTIVE
                 existing.state_version += 1
                 existing.updated_at = now
             return existing, occ
@@ -189,6 +209,7 @@ class InventoryService:
             moved_from.current_path = path_str
             moved_from.byte_size = size
             moved_from.missing_at = None
+            moved_from.availability_state = availability.ACTIVE
             moved_from.state_version += 1
             moved_from.updated_at = now
             self._open_path(session, moved_from.id, path_str, now, Occurrence.MOVED.value)

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():

photo_pipeline/services/restores.py

@@ -0,0 +1,648 @@
+"""RestoreService — plan and execute safe restores (US06-04).
+
+Restore is archiving read backwards, with one decisive difference: it removes
+nothing. The archived copy stays on its medium, so every failure mode here costs
+at most a discarded temporary file. What restore must never do is *lose identity*
+— the asset that comes back is the same asset, with its duplicate decision, safety
+review, analysis, and upload history intact — or *overwrite* something in the
+active library.
+
+Preflight proves, per concept §9 "Restore":
+
+- the recorded medium is mounted and is the right one (marker ``media_id``);
+- every selected asset is archived, its archive copy exists, and it hashes to
+  exactly the bytes the database recorded — a mismatch is ``divergent`` and is
+  refused, never silently accepted as "the file";
+- the destination lies inside the library, outside ``_IGNORE/``, and is free; a
+  taken path is answered with a collision-free name, never an overwrite;
+- the library filesystem has room for the scope plus the configured reserve;
+- no rename, archive, or restore lease is holding the lane.
+
+Blocker codes: ``no_library_root``, ``location_offline``, ``wrong_volume``,
+``unsafe_destination``, ``library_not_writable``, ``insufficient_capacity``,
+``lock_conflict``, ``rename_pending``, ``archive_pending``, ``empty_scope``,
+``not_archived``, ``archive_missing``, ``bytes_changed``.
+
+Per item the sequence is:
+
+```
+journal.begin (transferring)   ← intent persisted BEFORE any disk change
+recheck: medium, hash, free destination, asset still archived
+copy to a temporary file beside the destination, fsync, hash it back
+atomically publish into the library
+journal → verified
+current_path = destination, availability = active, path occurrence opened
+journal → complete
+```
+
+Like archiving, the confirmation token is derived from the report, so a changed
+scope, a swapped medium, or a destination that filled up invalidates it.
+"""
+
+from __future__ import annotations
+
+import hashlib
+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.jobs.domain_handlers import ARCHIVE_LOCK, LIBRARY_WRITE_LOCK, UPLOAD_LOCK
+from photo_pipeline.models import ArchiveLocation, ArchiveOperation, ArchivePlan, Asset, AssetPath
+from photo_pipeline.path_policy import PathPolicyError, is_excluded, normalize_root, resolve_within
+from photo_pipeline.services import availability
+from photo_pipeline.services.archive_journal import (
+    MANUAL,
+    RESTORE,
+    RESUMABLE,
+    ArchiveJournal,
+    ArchiveState,
+)
+from photo_pipeline.services.archive_transfer import (
+    _clean_temp_files,
+    _fsync_dir,
+    _plan_dict,
+    copy_verify_publish,
+)
+from photo_pipeline.services.archives import ArchiveError
+from photo_pipeline.services.hashing import sha256_file
+from photo_pipeline.services.jobs import JobService
+from photo_pipeline.services.rename_apply import PreconditionFailed, maybe_fault
+from photo_pipeline.services.rename_journal import RenameJournal
+
+PREFLIGHT_VERSION = 1
+TOKEN_PREFIX = f"r{PREFLIGHT_VERSION}"
+# What a restored file is called when its original name is taken. The suffix is
+# visible on purpose: a restore that quietly reuses a name is indistinguishable
+# from an overwrite.
+RESTORED_SUFFIX = "restored"
+
+LOCKS = (LIBRARY_WRITE_LOCK, UPLOAD_LOCK, ARCHIVE_LOCK)
+
+APPLYABLE_PLAN_STATES = frozenset({"planned", "applying", "failed", "complete"})
+
+
+def _now() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+def _issue(code: str, message: str) -> dict:
+    return {"code": code, "message": message}
+
+
+class RestoreService:
+    def __init__(self, session_factory: sessionmaker, *, config: Config) -> None:
+        self._session_factory = session_factory
+        self._config = config
+        self._roots = tuple(normalize_root(root) for root in config.library_roots)
+        self.journal = ArchiveJournal(session_factory)
+
+    # ── preflight ─────────────────────────────────────────────────────────────
+
+    def preflight(self, location_id: str, asset_ids: list[str] | None = None) -> dict:
+        """Validate a restore scope and issue its token. Nothing is written."""
+        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}")
+            root = Path(location.root)
+            online = availability.location_online(location)
+            marker = availability.read_marker(root)
+            report = {
+                "schema_version": PREFLIGHT_VERSION,
+                "location": {
+                    "id": location.id,
+                    "name": location.name,
+                    "root": str(root),
+                    "media_id": location.media_id,
+                    "state": _location_state(root, marker, location.media_id),
+                },
+                "blockers": [],
+            }
+            items = self._items(session, location, asset_ids, reachable=online)
+
+        report["blockers"] += self._destination_blockers(report["location"]["state"], root)
+        report["blockers"] += self._lock_blockers()
+        report["items"] = items
+        report["totals"] = {
+            "assets": len(items),
+            "blocked": sum(1 for item in items if item["blockers"]),
+            "bytes": sum(item["byte_size"] or 0 for item in items),
+        }
+        report["capacity"] = self._capacity(report["totals"]["bytes"])
+        if not report["capacity"]["sufficient"]:
+            report["blockers"].append(
+                _issue(
+                    "insufficient_capacity",
+                    f"{report['totals']['bytes']} B plus a "
+                    f"{self._config.archive_free_space_reserve_bytes} B reserve do not fit in "
+                    f"{report['capacity']['free_bytes']} B of free space",
+                )
+            )
+        if not items:
+            report["blockers"].append(
+                _issue("empty_scope", "no archived assets are in the selected scope")
+            )
+        report["state"] = (
+            "ready"
+            if not report["blockers"] and not report["totals"]["blocked"]
+            else "blocked"
+        )
+        report["token"] = _token(report)
+        report["generated_at"] = _now().isoformat()
+        return report
+
+    def verify_token(self, token: str, location_id: str, asset_ids: list[str] | None = None) -> bool:
+        return bool(token) and token == self.preflight(location_id, asset_ids)["token"]
+
+    def _items(
+        self, session, location: ArchiveLocation, asset_ids: list[str] | None, *, reachable: bool
+    ) -> list[dict]:
+        stmt = select(Asset).where(Asset.archive_location_id == location.id)
+        if asset_ids is None:
+            # A restored asset keeps its archive link; the default scope is only what
+            # is still archived, so restoring twice is an empty scope, not a blocker.
+            stmt = stmt.where(Asset.availability_state.in_(availability.ARCHIVED))
+        else:
+            stmt = stmt.where(Asset.id.in_(asset_ids))
+        assets = list(session.scalars(stmt.order_by(Asset.archive_path)))
+        if asset_ids is not None:
+            unknown = sorted(set(asset_ids) - {asset.id for asset in assets})
+            if unknown:
+                raise ArchiveError(
+                    "unknown_asset", f"not archived at this location: {', '.join(unknown)}"
+                )
+        taken: set[str] = set()
+        return [self._item(asset, location, reachable=reachable, taken=taken) for asset in assets]
+
+    def _item(self, asset: Asset, location: ArchiveLocation, *, reachable: bool, taken: set) -> dict:
+        source = Path(location.root) / (asset.archive_path or "")
+        blockers: list[dict] = []
+        archive_sha256 = None
+
+        if asset.availability_state not in availability.ARCHIVED:
+            blockers.append(
+                _issue("not_archived", f"asset {asset.id} is {asset.availability_state}")
+            )
+        if reachable:
+            if not source.exists():
+                blockers.append(_issue("archive_missing", f"{source} is not on the medium"))
+            else:
+                archive_sha256 = sha256_file(source)
+                if asset.current_sha256 and archive_sha256 != asset.current_sha256:
+                    blockers.append(
+                        _issue(
+                            "bytes_changed",
+                            f"{source} holds bytes that are not the recorded ones; "
+                            "the archived copy is divergent",
+                        )
+                    )
+
+        destination, destination_blockers = self._destination(asset, taken)
+        blockers += destination_blockers
+        if destination is not None:
+            taken.add(str(destination))
+        return {
+            "asset_id": asset.id,
+            "archive_path": asset.archive_path,
+            "source_path": str(source),
+            "destination_path": str(destination) if destination else None,
+            "expected_sha256": asset.current_sha256,
+            "archive_sha256": archive_sha256,
+            "byte_size": asset.byte_size,
+            "availability_state": asset.availability_state,
+            "blockers": blockers,
+        }
+
+    def _destination(self, asset: Asset, taken: set) -> tuple[Path | None, list[dict]]:
+        """A free path inside the library that mirrors the archived layout.
+
+        Restoring onto an existing file is never an option, so a taken name is
+        answered with ``name (restored).ext`` — visible, ordinary, and impossible to
+        confuse with an overwrite.
+        """
+        if not self._roots:
+            return None, [_issue("no_library_root", "no library root is configured")]
+        root = self._roots[0]
+        try:
+            candidate = resolve_within(root, root / (asset.archive_path or ""))
+        except PathPolicyError as error:
+            return None, [_issue("unsafe_destination", str(error))]
+        if is_excluded(candidate):
+            return None, [
+                _issue("unsafe_destination", f"{candidate} is inside an excluded (_IGNORE/) tree")
+            ]
+        return _free_path(candidate, taken), []
+
+    def _destination_blockers(self, state: str, root: Path) -> list[dict]:
+        blockers: list[dict] = []
+        if not self._roots:
+            blockers.append(_issue("no_library_root", "no library root is configured"))
+        elif not os.access(self._roots[0], os.W_OK):
+            blockers.append(
+                _issue("library_not_writable", f"{self._roots[0]} is not writable")
+            )
+        if state == "offline":
+            blockers.append(
+                _issue("location_offline", f"the archive medium is not mounted at {root}")
+            )
+        elif state == "wrong_volume":
+            blockers.append(_issue("wrong_volume", f"{root} holds a different archive medium"))
+        return blockers
+
+    def _lock_blockers(self) -> list[dict]:
+        blockers: list[dict] = []
+        jobs = JobService(self._session_factory)
+        for lock in LOCKS:
+            held = jobs.blockers(lock)
+            if held:
+                blockers.append(
+                    _issue("lock_conflict", f"the {lock} lane is busy: job {held[0]['id']}")
+                )
+        if RenameJournal(self._session_factory).blocks_mutation():
+            blockers.append(
+                _issue("rename_pending", "an unresolved rename must be recovered before restoring")
+            )
+        if self.journal.blocks_mutation():
+            blockers.append(
+                _issue(
+                    "archive_pending",
+                    "an unresolved archive or restore must be recovered before restoring",
+                )
+            )
+        return blockers
+
+    def _capacity(self, required: int) -> dict:
+        reserve = self._config.archive_free_space_reserve_bytes
+        free = shutil.disk_usage(self._roots[0]).free if self._roots else None
+        return {
+            "required_bytes": required,
+            "reserve_bytes": reserve,
+            "free_bytes": free,
+            "sufficient": free is not None and free >= required + reserve,
+        }
+
+    # ── plans ─────────────────────────────────────────────────────────────────
+
+    def create(self, location_id: str, asset_ids: list[str] | None = None, *, token: str) -> dict:
+        preflight = self.preflight(location_id, asset_ids)
+        if not token or token != preflight["token"]:
+            raise ArchiveError("stale_token", "the restore preflight changed since it was approved")
+        if preflight["state"] != "ready":
+            codes = ", ".join(sorted({issue["code"] for issue in preflight["blockers"]})) or "-"
+            blocked = sorted(
+                {issue["code"] for item in preflight["items"] for issue in item["blockers"]}
+            )
+            raise ArchiveError(
+                "blocked", f"the restore scope is blocked: {', '.join(blocked) or codes}"
+            )
+
+        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(asset_ids) if asset_ids is not None else None,
+                    direction=RESTORE,
+                    state="planned",
+                    schema_version=PREFLIGHT_VERSION,
+                    asset_count=preflight["totals"]["assets"],
+                    byte_size=preflight["totals"]["bytes"],
+                )
+            )
+            session.flush()
+            for sequence, item in enumerate(preflight["items"]):
+                session.add(
+                    ArchiveOperation(
+                        id=str(uuid.uuid4()),
+                        plan_id=plan_id,
+                        direction=RESTORE,
+                        sequence=sequence,
+                        album=Path(item["archive_path"]).parent.name or "(root)",
+                        asset_id=item["asset_id"],
+                        source_path=item["source_path"],
+                        destination_path=item["destination_path"],
+                        archive_path=item["archive_path"],
+                        expected_sha256=item["expected_sha256"],
+                        byte_size=item["byte_size"],
+                        journal_state=ArchiveState.PLANNED,
+                    )
+                )
+            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 or plan.direction != RESTORE:
+                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)
+                .where(ArchivePlan.direction == RESTORE)
+                .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 = "restore"
+    ) -> dict:
+        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']}")
+        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"])
+        restored = failed = skipped = 0
+        for operation in self.journal.operations(plan_id):
+            if operation["journal_state"] == ArchiveState.COMPLETE:
+                skipped += 1
+                continue
+            try:
+                if operation["journal_state"] == ArchiveState.VERIFIED:
+                    self._finish(operation, token=token)
+                else:
+                    self._restore_one(operation, location, token=token, worker_id=worker_id)
+                restored += 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, "restore_error", str(error))
+                failed += 1
+        state = self.journal.sync_plan_state(plan_id)
+        return {
+            "plan_id": plan_id,
+            "restored": restored,
+            "failed": failed,
+            "skipped": skipped,
+            "state": state,
+        }
+
+    def _restore_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; from here a crash is resolvable from journal + disk.
+        self.journal.begin(operation["id"], worker_id=worker_id, fencing_token=token)
+        maybe_fault(ArchiveState.TRANSFERRING)
+
+        # 2. Recheck against the medium and the library as they are right now.
+        self._recheck(operation, source, destination, location)
+        destination.parent.mkdir(parents=True, exist_ok=True)
+
+        # 3. Always copy: the archived original stays on its medium.
+        copy_verify_publish(source, destination, operation["expected_sha256"])
+        _fsync_dir(destination.parent)
+
+        if sha256_file(destination) != operation["expected_sha256"]:
+            raise PreconditionFailed(
+                "restore_mismatch", f"{destination} does not hold the expected bytes"
+            )
+        self.journal.transition(operation["id"], ArchiveState.VERIFIED, fencing_token=token)
+        maybe_fault(ArchiveState.VERIFIED)
+
+        self._finish(self.journal.get(operation["id"]), token=token)
+
+    def _finish(self, operation: dict, *, token: int) -> None:
+        """Publish the restored file to the database. Idempotent, so recovery may
+        replay it after a crash between the copy and the bookkeeping."""
+        destination = Path(operation["destination_path"])
+        if not destination.exists() or sha256_file(destination) != operation["expected_sha256"]:
+            raise PreconditionFailed(
+                "restore_unverified", f"{destination} is not a verified restored copy"
+            )
+        self._record_restored(operation, 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:
+        root = Path(location["root"])
+        if not root.is_dir() or not (root / availability.MARKER_NAME).exists():
+            raise PreconditionFailed("location_offline", f"{root} is not the archive medium")
+        if not source.exists():
+            raise PreconditionFailed("archive_missing", f"{source} is not on the medium")
+        if source.is_symlink() or destination.is_symlink():
+            raise PreconditionFailed("symlink", "refusing to restore through a symlink")
+        if destination.exists():
+            # Never overwrite: the plan's free path was taken since it was made.
+            raise PreconditionFailed(
+                "destination_exists", f"destination {destination} is occupied"
+            )
+        if not self._inside_library(destination):
+            raise PreconditionFailed(
+                "destination_escape", f"{destination} is outside the library roots"
+            )
+        if sha256_file(source) != operation["expected_sha256"]:
+            self._mark_divergent(operation["asset_id"])
+            raise PreconditionFailed(
+                "bytes_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.availability_state not in availability.ARCHIVED:
+                raise PreconditionFailed(
+                    "not_archived", f"asset {operation['asset_id']} is no longer archived"
+                )
+
+    def _inside_library(self, destination: Path) -> bool:
+        for root in self._roots:
+            try:
+                resolve_within(root, destination)
+                return True
+            except PathPolicyError:
+                continue
+        return False
+
+    # ── database ──────────────────────────────────────────────────────────────
+
+    def _record_restored(self, operation: dict, destination: Path) -> None:
+        """The bytes are back in the library: open the new active occurrence and set
+        availability. Identity, decisions, and history are untouched — that is the
+        entire point of restoring rather than re-importing."""
+        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"
+                )
+            # A restored asset may be returning to a path it once held, so only an
+            # *open* occurrence counts as already registered — that is what keeps
+            # recovery idempotent without collapsing the path history.
+            recorded = session.scalar(
+                select(AssetPath).where(
+                    AssetPath.asset_id == asset.id,
+                    AssetPath.path == str(destination),
+                    AssetPath.valid_until.is_(None),
+                )
+            )
+            if recorded is None:  # idempotent: recovery may replay this
+                session.add(
+                    AssetPath(
+                        asset_id=asset.id,
+                        path=str(destination),
+                        valid_from=now,
+                        reason="restore",
+                    )
+                )
+            asset.current_path = str(destination)
+            asset.availability_state = availability.ACTIVE
+            asset.missing_at = None
+            # The archive copy stays where it is; keeping the link means a restored
+            # asset still knows which medium holds its archived bytes.
+            asset.archive_divergent_at = None
+            asset.state_version += 1
+            asset.updated_at = now
+            session.commit()
+
+    def _mark_divergent(self, asset_id: str) -> None:
+        """Record that the archived copy is not the recorded file. Durable, because
+        the next restore attempt must not rediscover this from scratch."""
+        with self._session_factory() as session:
+            asset = session.get(Asset, asset_id)
+            if asset is None:
+                return
+            asset.archive_divergent_at = _now()
+            asset.state_version += 1
+            session.commit()
+
+    # ── recovery ──────────────────────────────────────────────────────────────
+
+    def recover(self, *, worker_id: str = "restore-recovery") -> dict:
+        """Resolve every incomplete restore from journal + disk evidence.
+
+        A restore never removed anything, so ``resumable`` simply discards the
+        temporary debris and re-plans the item; ``forward`` finishes the bookkeeping
+        for a published file; ``manual`` is left untouched and keeps blocking.
+        """
+        results = {"resumed": 0, "completed": 0, "manual": 0}
+        touched: set[str] = set()
+        for verdict in self.journal.classify_all(direction=RESTORE):
+            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:
+                _clean_temp_files(Path(operation["destination_path"]).parent)
+                self.journal.transition(operation["id"], ArchiveState.PLANNED, fencing_token=token)
+                results["resumed"] += 1
+                continue
+            try:
+                self._finish(operation, token=token)
+                results["completed"] += 1
+            except PreconditionFailed as error:
+                self._fail(operation, token, error.code, str(error))
+                results["manual"] += 1
+        for plan_id in touched:
+            self.journal.sync_plan_state(plan_id)
+        return results
+
+    def recovery_status(self) -> dict:
+        verdicts = self.journal.classify_all(direction=RESTORE)
+        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:
+        plan = self.get(plan_id)
+        if plan is None:
+            raise ArchiveError("unknown_plan", f"unknown restore plan {plan_id!r}")
+        return 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:
+        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 _location_state(root: Path, marker: dict | None, media_id: str) -> str:
+    if not root.is_dir() or marker is None:
+        return "offline"
+    return "online" if marker.get("media_id") == media_id else "wrong_volume"
+
+
+def _free_path(candidate: Path, taken: set) -> Path:
+    """``a.jpg`` → ``a (restored).jpg`` → ``a (restored 2).jpg`` …
+
+    ``taken`` holds the destinations already claimed by earlier items of the same
+    plan, so two restores in one scope cannot plan the same path.
+    """
+    if not candidate.exists() and str(candidate) not in taken:
+        return candidate
+    stem, suffix = candidate.stem, candidate.suffix
+    attempt = 1
+    while True:
+        label = RESTORED_SUFFIX if attempt == 1 else f"{RESTORED_SUFFIX} {attempt}"
+        alternative = candidate.with_name(f"{stem} ({label}){suffix}")
+        if not alternative.exists() and str(alternative) not in taken:
+            return alternative
+        attempt += 1
+
+
+def _token(report: dict) -> str:
+    """Digest of everything the report asserts about the scope and the medium.
+
+    Free space is excluded: it drifts constantly without changing what a restore
+    would do, and the capacity verdict itself is part of the digest.
+    """
+    payload = {key: value for key, value in report.items() if key not in ("generated_at", "token")}
+    payload["capacity"] = {
+        key: value for key, value in payload["capacity"].items() if key != "free_bytes"
+    }
+    digest = hashlib.sha256(
+        json.dumps(payload, sort_keys=True, ensure_ascii=False, default=str).encode("utf-8")
+    ).hexdigest()
+    return f"{TOKEN_PREFIX}:{digest}"

photo_pipeline/services/thumbnails.py

@@ -8,6 +8,11 @@ an EXIF-only edit reuses the file while a real pixel change invalidates it; writ
 are atomic and the cache is bounded by an LRU quota. Failures are persisted as typed
 errors so a broken original is not retried on every request.
 
+An archived asset is served from its medium when that medium is mounted, and from
+its *protected* preview when it is not (US06-03). Protected previews are evidence,
+not cache: the quota never evicts them, because the original they describe may be
+unreachable when duplicate review needs it.
+
 Reuses photo_analyzer.prepare_image decode/resize/HEIC handling, adding the missing
 EXIF-orientation step, WebP output, and a managed cache (donor_ledger.yaml:
 pa-imaging).
@@ -26,6 +31,7 @@ from sqlalchemy.orm import sessionmaker
 from photo_pipeline import path_policy
 from photo_pipeline.config import Config
 from photo_pipeline.models import Asset, Thumbnail
+from photo_pipeline.services import availability
 
 # Best-effort HEIC support: registered only if the optional decoder is installed.
 try:  # pragma: no cover - depends on an optional native dependency
@@ -38,6 +44,8 @@ except Exception:  # pragma: no cover
 SIZES = (256, 512, 1280)
 THUMB_VERSION = 1
 THUMB_FORMAT = "webp"
+# The size kept as durable comparison evidence for archived assets (concept §9).
+PROTECTED_SIZE = 1280
 
 
 class ThumbnailError(RuntimeError):
@@ -86,7 +94,12 @@ class ThumbnailService:
         self._config = config
         self._cache_dir = config.thumbnail_cache_dir
 
-    def generate(self, asset_id: str, size: int) -> Path:
+    def generate(
+        self, asset_id: str, size: int, *, protected: bool = False, source: Path | None = None
+    ) -> Path:
+        """Render (or reuse) a preview. ``source`` overrides where the bytes are read
+        from — the archiver passes its verified archive copy, which the database does
+        not yet point at while the transfer is still in flight."""
         if size not in SIZES:
             raise InvalidSize(f"size must be one of {SIZES}")
 
@@ -94,9 +107,7 @@ class ThumbnailService:
             asset = session.get(Asset, asset_id)
             if asset is None:
                 raise ThumbnailNotFound(f"unknown asset {asset_id}")
-            if asset.availability_state != "active" or not asset.current_path:
-                raise ThumbnailUnavailable(f"asset {asset_id} has no active file")
-            self._validate_path(asset.current_path)
+            archived = asset.availability_state in availability.ARCHIVED
             cache_key = self._cache_key(asset, size)
 
             row = session.get(Thumbnail, cache_key)
@@ -107,9 +118,18 @@ class ThumbnailService:
                     )
                 if row.path and Path(row.path).exists():
                     _touch(row.path)
+                    if protected and not row.protected:
+                        self._protect(cache_key)
                     return Path(row.path)
 
-            source = asset.current_path
+            # An archived original is read from its medium; when that medium is not
+            # mounted the retained preview above is the only evidence there is.
+            source = source or availability.readable_path(session, asset)
+            if source is None:
+                raise ThumbnailUnavailable(f"asset {asset_id} has no readable file")
+            source = str(source)
+            if source == asset.current_path:
+                self._validate_path(source)  # archive roots lie outside the library
 
         # Rendering happens outside the DB session (no transaction held during I/O).
         try:
@@ -119,10 +139,61 @@ class ThumbnailService:
                 self._record_error(cache_key, asset_id, size, error.code)
             raise
 
-        self._record_ready(cache_key, asset_id, size, rendered)
+        # Archived assets keep their preview permanently: it is the comparison
+        # evidence that survives the original leaving active storage.
+        self._record_ready(cache_key, asset_id, size, rendered, protected=protected or archived)
         self._enforce_quota(keep=rendered["path"])
         return Path(rendered["path"])
 
+    def ensure_protected(self, asset_id: str, *, source: Path | None = None) -> dict:
+        """Produce (or confirm) the durable comparison preview for an asset.
+
+        Returns evidence rather than raising, because the caller — archive
+        preflight and the transfer itself — decides what an unrenderable original
+        means. ``unsupported`` is a recorded property of the file, not a failure of
+        the policy: its hashes and metadata remain the comparison evidence.
+        """
+        try:
+            path = self.generate(asset_id, PROTECTED_SIZE, protected=True, source=source)
+        except tuple(_PERSISTED_ERRORS) as error:
+            return {"state": "unsupported", "error_code": error.code, "path": None}
+        except ThumbnailError as error:
+            return {"state": "unavailable", "error_code": error.code, "path": None}
+        return {"state": "ready", "error_code": None, "path": str(path)}
+
+    def evidence(self, asset_id: str) -> dict:
+        """What durable preview this asset has right now, without rendering."""
+        with self._session_factory() as session:
+            rows = list(
+                session.execute(
+                    select(Thumbnail).where(Thumbnail.asset_id == asset_id)
+                ).scalars()
+            )
+        for row in rows:
+            if row.state == "ready" and row.path and Path(row.path).exists():
+                return {
+                    "state": "ready",
+                    "protected": bool(row.protected),
+                    "size": row.size,
+                    "error_code": None,
+                }
+        for row in rows:
+            if row.state == "error":
+                return {
+                    "state": "unsupported",
+                    "protected": False,
+                    "size": row.size,
+                    "error_code": row.error_code,
+                }
+        return {"state": "missing", "protected": False, "size": None, "error_code": None}
+
+    def _protect(self, cache_key: str) -> None:
+        with self._session_factory() as session:
+            row = session.get(Thumbnail, cache_key)
+            if row is not None:
+                row.protected = True
+                session.commit()
+
     # ── path safety ──────────────────────────────────────────────────────────
     def _validate_path(self, current_path: str) -> None:
         path = Path(current_path)
@@ -184,7 +255,9 @@ class ThumbnailService:
         }
 
     # ── persistence ────────────────────────────────────────────────────────────
-    def _record_ready(self, cache_key: str, asset_id: str, size: int, rendered: dict) -> None:
+    def _record_ready(
+        self, cache_key: str, asset_id: str, size: int, rendered: dict, *, protected: bool = False
+    ) -> None:
         with self._session_factory() as session:
             session.merge(
                 Thumbnail(
@@ -197,6 +270,7 @@ class ThumbnailService:
                     width=rendered["width"],
                     height=rendered["height"],
                     format=rendered["format"],
+                    protected=protected,
                 )
             )
             try:
@@ -230,6 +304,10 @@ class ThumbnailService:
         if total <= quota:
             return
         files.sort(key=lambda f: f.stat().st_mtime)  # least-recently-used first
+        # Protected previews are evidence, not cache: an archived original cannot be
+        # re-rendered once its medium is away, so eviction never touches them.
+        protected = self._protected_paths()
+        files = [f for f in files if str(f) not in protected]
         keep_path = str(Path(keep)) if keep else None
         evicted: list[str] = []
         for f in files:
@@ -247,6 +325,16 @@ class ThumbnailService:
         if evicted:
             self._forget(evicted)
 
+    def _protected_paths(self) -> set[str]:
+        with self._session_factory() as session:
+            return {
+                row.path
+                for row in session.execute(
+                    select(Thumbnail).where(Thumbnail.protected.is_(True))
+                ).scalars()
+                if row.path
+            }
+
     def _forget(self, paths: list[str]) -> None:
         with self._session_factory() as session:
             rows = session.execute(

pyproject.toml

@@ -32,4 +32,5 @@ markers = [
     "phase_c: Phase C end-to-end acceptance (US03-05) — album proposal API and browser journeys",
     "phase_d: Phase D end-to-end acceptance (US04-06) — guarded rename API, fault, and browser journeys",
     "phase_e: Phase E end-to-end acceptance (US05-06) — upload preflight, uploader, and browser journeys",
+    "phase_f: Phase F end-to-end acceptance (US06-06) — archive destination, transfer, and restore journeys",
 ]

tests/integration/test_archive_preflight.py

@@ -0,0 +1,592 @@
+"""Archive destinations and preflight (US06-01).
+
+Archive is the only stage that removes originals, so every case here asks the same
+question: would this preflight let an album leave active storage when it should
+not? The destinations are real directories on real filesystems — mounted, missing,
+swapped for another medium, read-only, or full — and preflight itself must stay
+non-destructive: the library snapshot is asserted unchanged.
+"""
+
+import json
+import os
+import stat
+import uuid
+from datetime import datetime, timedelta, timezone
+
+import pytest
+from fastapi.testclient import TestClient
+
+from photo_pipeline.api.app import create_app
+from photo_pipeline.config import Config
+from photo_pipeline.db import create_db_engine, create_session_factory, run_migrations
+from photo_pipeline.jobs.domain_handlers import ARCHIVE_LOCK, UPLOAD_LOCK
+from photo_pipeline.models import (
+    Asset,
+    RenameOperation,
+    RenamePlan,
+    UploadBatch,
+    UploadItem,
+)
+from photo_pipeline.services.archives import MARKER_NAME, ArchiveError, ArchiveService
+from photo_pipeline.services.hashing import sha256_file
+from photo_pipeline.services.jobs import JobService
+
+pytestmark = pytest.mark.phase_f  # part of the Phase F acceptance gate (US06-06)
+
+NOW = datetime(2026, 1, 1, tzinfo=timezone.utc)
+
+
+# ── environment ──────────────────────────────────────────────────────────────
+
+
+def _env(tmp_path, *, reserve=0):
+    (tmp_path / "data").mkdir(exist_ok=True)
+    lib = tmp_path / "lib"
+    lib.mkdir(exist_ok=True)
+    archive = tmp_path / "archive"
+    archive.mkdir(exist_ok=True)
+    config = Config.from_env(
+        {
+            "PHOTO_PIPELINE_DATA_DIR": str(tmp_path / "data"),
+            "PHOTO_PIPELINE_LIBRARY_ROOTS": str(lib),
+            "PHOTO_PIPELINE_ARCHIVE_FREE_SPACE_RESERVE_BYTES": str(reserve),
+        }
+    )
+    run_migrations(config.database_url)
+    return config, create_session_factory(create_db_engine(config.database_url)), lib, archive
+
+
+def _album(
+    sf,
+    lib,
+    album="rome",
+    names=("a.jpg", "b.jpg"),
+    *,
+    uploaded=True,
+    outcome="uploaded",
+    outcome_state="verified",
+    stale_bytes=False,
+):
+    """A real album folder whose assets carry their upload evidence."""
+    folder = lib / album
+    folder.mkdir(parents=True, exist_ok=True)
+    ids = []
+    with sf() as session:
+        batch_id = str(uuid.uuid4())
+        if uploaded:
+            session.add(
+                UploadBatch(
+                    id=batch_id,
+                    album=album,
+                    folder=str(folder),
+                    album_name=album,
+                    state="succeeded",
+                    preflight_token="v1:test",
+                    outcome_state=outcome_state,
+                    stale_bytes=stale_bytes,
+                    created_at=NOW,
+                )
+            )
+        for name in names:
+            path = folder / name
+            path.write_bytes(name.encode() * 16)
+            asset_id = str(uuid.uuid4())
+            ids.append(asset_id)
+            session.add(
+                Asset(
+                    id=asset_id,
+                    original_path=str(path),
+                    current_path=str(path),
+                    discovered_at=NOW,
+                    hash_version=1,
+                    byte_size=path.stat().st_size,
+                    current_sha256=sha256_file(path),
+                )
+            )
+            if uploaded:
+                session.add(
+                    UploadItem(
+                        batch_id=batch_id,
+                        asset_id=asset_id,
+                        path=str(path),
+                        sha256=sha256_file(path),
+                        sha1="0" * 40,
+                        state="sent",
+                        outcome=outcome,
+                    )
+                )
+        session.commit()
+    return folder, ids
+
+
+def _service(sf, config):
+    return ArchiveService(sf, config=config)
+
+
+def _location(sf, config, archive, name="external"):
+    return _service(sf, config).register(name, str(archive))
+
+
+def _snapshot(lib):
+    return {
+        str(p.relative_to(lib)): (p.read_bytes() if p.is_file() else None)
+        for p in sorted(lib.rglob("*"))
+    }
+
+
+def _codes(report):
+    return (
+        {issue["code"] for issue in report["blockers"]}
+        | {issue["code"] for album in report["albums"] for issue in album["blockers"]}
+        | {
+            issue["code"]
+            for album in report["albums"]
+            for asset in album["assets"]
+            for issue in asset["blockers"]
+        }
+    )
+
+
+# ── locations ────────────────────────────────────────────────────────────────
+
+
+def test_registering_a_location_stamps_the_medium_with_its_identity(tmp_path):
+    config, sf, _, archive = _env(tmp_path)
+
+    location = _location(sf, config, archive)
+
+    marker = json.loads((archive / MARKER_NAME).read_text())
+    assert marker["media_id"] == location["media_id"]
+    assert location["state"] == "online" and location["writable"] is True
+    assert location["root"] == str(archive.resolve())
+    listed = _service(sf, config).locations()
+    assert [(row["id"], row["media_id"], row["state"]) for row in listed] == [
+        (location["id"], location["media_id"], "online")
+    ]
+
+
+def test_a_second_location_cannot_claim_the_same_medium(tmp_path):
+    config, sf, _, archive = _env(tmp_path)
+    _location(sf, config, archive)
+
+    with pytest.raises(ArchiveError) as error:
+        _location(sf, config, archive, name="second")
+
+    assert error.value.code == "already_registered"
+
+
+@pytest.mark.parametrize("inside", ["", "sub"])
+def test_a_destination_inside_the_library_is_refused(tmp_path, inside):
+    """The library may never archive into itself: the 'reclaimed' bytes would still
+    be in the active tree, and a later scan would rediscover them."""
+    config, sf, lib, _ = _env(tmp_path)
+    root = lib / inside if inside else lib
+    root.mkdir(exist_ok=True)
+
+    with pytest.raises(ArchiveError) as error:
+        _service(sf, config).register("bad", str(root))
+
+    assert error.value.code == "unsafe_destination"
+
+
+def test_an_ignored_destination_is_refused(tmp_path):
+    config, sf, _, _ = _env(tmp_path)
+    root = tmp_path / "_IGNORE" / "archive"
+    root.mkdir(parents=True)
+
+    with pytest.raises(ArchiveError) as error:
+        _service(sf, config).register("ignored", str(root))
+
+    assert error.value.code == "unsafe_destination"
+
+
+def test_listing_reports_an_unmounted_medium_as_offline(tmp_path):
+    config, sf, _, archive = _env(tmp_path)
+    _location(sf, config, archive)
+    (archive / MARKER_NAME).unlink()
+
+    assert [row["state"] for row in _service(sf, config).locations()] == ["offline"]
+
+
+# ── happy path ───────────────────────────────────────────────────────────────
+
+
+def test_ready_preflight_previews_scope_method_and_reclaimable_bytes(tmp_path):
+    config, sf, lib, archive = _env(tmp_path)
+    folder, ids = _album(sf, lib)
+    location = _location(sf, config, archive)
+    before = _snapshot(lib)
+
+    report = _service(sf, config).preflight(location["id"])
+
+    assert report["state"] == "ready" and report["blockers"] == []
+    album = report["albums"][0]
+    assert album["album"] == "rome" and album["folder"] == str(folder)
+    assert album["destination"] == str(archive.resolve() / "rome")
+    assert album["transfer_method"] in ("move", "copy_verify_remove")
+    assert album["reclaimable_bytes"] == sum(p.stat().st_size for p in folder.iterdir())
+    assert sorted(a["asset_id"] for a in album["assets"]) == sorted(ids)
+    assert report["totals"]["bytes"] == album["reclaimable_bytes"]
+    assert report["capacity"]["sufficient"] is True
+    # Both must be proven by writing, not assumed.
+    assert report["backup"]["ok"] is True and report["backup"]["bytes"] > 0
+    assert report["manifest"]["ok"] is True
+    assert report["token"].startswith("v1:")
+    assert _snapshot(lib) == before, "preflight must not touch the library"
+    assert not list(archive.glob("*probe*")), "probe files must be cleaned up"
+
+
+def test_same_filesystem_destination_is_previewed_as_a_move(tmp_path):
+    config, sf, lib, archive = _env(tmp_path)
+    _album(sf, lib)
+    location = _location(sf, config, archive)
+
+    report = _service(sf, config).preflight(location["id"])
+
+    # tmp_path is one filesystem, so this is the same-filesystem case by construction.
+    assert report["albums"][0]["transfer_method"] == "move"
+
+
+def test_scoping_to_one_album_excludes_the_others(tmp_path):
+    config, sf, lib, archive = _env(tmp_path)
+    _album(sf, lib, "rome")
+    _album(sf, lib, "paris", names=("c.jpg",))
+    location = _location(sf, config, archive)
+
+    report = _service(sf, config).preflight(location["id"], ["paris"])
+
+    assert [album["album"] for album in report["albums"]] == ["paris"]
+    assert report["totals"]["assets"] == 1
+
+
+def test_unknown_album_and_unknown_location_are_refused(tmp_path):
+    config, sf, lib, archive = _env(tmp_path)
+    _album(sf, lib)
+    location = _location(sf, config, archive)
+    service = _service(sf, config)
+
+    with pytest.raises(ArchiveError) as unknown_album:
+        service.preflight(location["id"], ["atlantis"])
+    with pytest.raises(ArchiveError) as unknown_location:
+        service.preflight("nope")
+
+    assert unknown_album.value.code == "unknown_album"
+    assert unknown_location.value.code == "unknown_location"
+
+
+def test_empty_scope_is_a_blocker(tmp_path):
+    config, sf, _, archive = _env(tmp_path)
+    location = _location(sf, config, archive)
+
+    report = _service(sf, config).preflight(location["id"])
+
+    assert report["state"] == "blocked" and "empty_scope" in _codes(report)
+
+
+# ── destination ──────────────────────────────────────────────────────────────
+
+
+def test_offline_medium_blocks(tmp_path):
+    config, sf, lib, archive = _env(tmp_path)
+    _album(sf, lib)
+    location = _location(sf, config, archive)
+    (archive / MARKER_NAME).unlink()  # the disk went away
+
+    report = _service(sf, config).preflight(location["id"])
+
+    assert report["state"] == "blocked" and "location_offline" in _codes(report)
+    assert report["location"]["state"] == "offline"
+
+
+def test_a_different_medium_at_the_same_mountpoint_blocks(tmp_path):
+    """The mountpoint is right, the disk is not — never write the archive here."""
+    config, sf, lib, archive = _env(tmp_path)
+    _album(sf, lib)
+    location = _location(sf, config, archive)
+    (archive / MARKER_NAME).write_text(json.dumps({"media_id": "some-other-disk"}))
+
+    report = _service(sf, config).preflight(location["id"])
+
+    assert "wrong_volume" in _codes(report)
+    assert report["location"]["state"] == "wrong_volume"
+
+
+def test_read_only_destination_blocks_and_cannot_write_the_manifest(tmp_path):
+    config, sf, lib, archive = _env(tmp_path)
+    _album(sf, lib)
+    location = _location(sf, config, archive)
+    mode = archive.stat().st_mode
+    archive.chmod(mode & ~stat.S_IWUSR & ~stat.S_IWGRP & ~stat.S_IWOTH)
+    try:
+        report = _service(sf, config).preflight(location["id"])
+    finally:
+        archive.chmod(mode)
+
+    assert {"destination_not_writable", "manifest_unwritable"} <= _codes(report)
+    assert report["manifest"]["ok"] is False
+
+
+@pytest.mark.skipif(os.geteuid() == 0, reason="root ignores directory permissions")
+def test_read_only_destination_is_detected_by_a_real_write(tmp_path):
+    config, sf, lib, archive = _env(tmp_path)
+    _album(sf, lib)
+    location = _location(sf, config, archive)
+    mode = archive.stat().st_mode
+    archive.chmod(stat.S_IRUSR | stat.S_IXUSR)
+    try:
+        report = _service(sf, config).preflight(location["id"])
+    finally:
+        archive.chmod(mode)
+
+    assert report["location"]["writable"] is False
+
+
+def test_insufficient_capacity_blocks(tmp_path):
+    """The reserve is what stops an archive from filling its own destination."""
+    config, sf, lib, archive = _env(tmp_path, reserve=10**15)
+    _album(sf, lib)
+    location = _location(sf, config, archive)
+
+    report = _service(sf, config).preflight(location["id"])
+
+    assert report["state"] == "blocked" and "insufficient_capacity" in _codes(report)
+    assert report["capacity"]["sufficient"] is False
+    assert report["capacity"]["reserve_bytes"] == 10**15
+
+
+def test_an_occupied_destination_blocks_that_album(tmp_path):
+    config, sf, lib, archive = _env(tmp_path)
+    _album(sf, lib)
+    location = _location(sf, config, archive)
+    (archive / "rome").mkdir()
+    (archive / "rome" / "a.jpg").write_bytes(b"something already here")
+
+    report = _service(sf, config).preflight(location["id"])
+
+    assert "destination_collision" in _codes(report)
+    assert report["albums"][0]["state"] == "blocked"
+
+
+def test_a_destination_moved_into_the_library_blocks_even_though_it_registered(tmp_path):
+    """Registration validated the root once; preflight validates it again, because a
+    mountpoint can be moved after the fact."""
+    config, sf, lib, archive = _env(tmp_path)
+    _album(sf, lib)
+    location = _location(sf, config, archive)
+    with sf() as session:
+        from photo_pipeline.models import ArchiveLocation
+
+        session.get(ArchiveLocation, location["id"]).root = str(lib / "inside")
+        session.commit()
+    (lib / "inside").mkdir()
+    (lib / "inside" / MARKER_NAME).write_text(json.dumps({"media_id": location["media_id"]}))
+
+    report = _service(sf, config).preflight(location["id"])
+
+    assert "unsafe_destination" in _codes(report)
+
+
+# ── source readiness ─────────────────────────────────────────────────────────
+
+
+@pytest.mark.parametrize(
+    "kwargs,code",
+    [
+        ({"uploaded": False}, "upload_unverified"),
+        ({"outcome_state": "requires_verification"}, "upload_unverified"),
+        ({"outcome": "failed"}, "upload_unverified"),
+        ({"outcome": "skipped"}, "upload_unverified"),
+        ({"stale_bytes": True}, "upload_unverified"),
+    ],
+)
+def test_an_unverified_upload_blocks_the_album(tmp_path, kwargs, code):
+    config, sf, lib, archive = _env(tmp_path)
+    _album(sf, lib, **kwargs)
+    location = _location(sf, config, archive)
+
+    report = _service(sf, config).preflight(location["id"])
+
+    assert report["state"] == "blocked" and code in _codes(report)
+    assert report["albums"][0]["blocked_count"] == 2
+
+
+@pytest.mark.parametrize("outcome", ["upgraded", "duplicate"])
+def test_upgraded_and_duplicate_uploads_are_evidence_enough(tmp_path, outcome):
+    """Immich already holds these exact bytes; that is what archiving requires."""
+    config, sf, lib, archive = _env(tmp_path)
+    _album(sf, lib, outcome=outcome)
+    location = _location(sf, config, archive)
+
+    assert _service(sf, config).preflight(location["id"])["state"] == "ready"
+
+
+def test_bytes_changed_since_upload_block_the_album(tmp_path):
+    config, sf, lib, archive = _env(tmp_path)
+    folder, _ = _album(sf, lib)
+    location = _location(sf, config, archive)
+    (folder / "a.jpg").write_bytes(b"edited after the upload")
+
+    report = _service(sf, config).preflight(location["id"])
+
+    assert {"bytes_changed", "partial_scope"} <= _codes(report)
+    assert report["albums"][0]["blocked_count"] == 1
+
+
+def test_a_missing_source_file_blocks_the_album(tmp_path):
+    config, sf, lib, archive = _env(tmp_path)
+    folder, _ = _album(sf, lib)
+    location = _location(sf, config, archive)
+    (folder / "a.jpg").unlink()
+
+    assert "file_missing" in _codes(_service(sf, config).preflight(location["id"]))
+
+
+# ── leases ───────────────────────────────────────────────────────────────────
+
+
+@pytest.mark.parametrize("lock", [UPLOAD_LOCK, ARCHIVE_LOCK])
+def test_a_held_lease_blocks_archiving(tmp_path, lock):
+    config, sf, lib, archive = _env(tmp_path)
+    _album(sf, lib)
+    location = _location(sf, config, archive)
+    JobService(sf).enqueue("upload_batch", lock=lock, items=["x"])
+
+    report = _service(sf, config).preflight(location["id"])
+
+    assert report["state"] == "blocked" and "lock_conflict" in _codes(report)
+
+
+def test_a_half_applied_rename_blocks_archiving(tmp_path):
+    config, sf, lib, archive = _env(tmp_path)
+    folder, _ = _album(sf, lib)
+    location = _location(sf, config, archive)
+    with sf() as session:
+        plan_id = str(uuid.uuid4())
+        session.add(RenamePlan(id=plan_id, state="applying", operation_count=1))
+        session.flush()
+        session.add(
+            RenameOperation(
+                id=str(uuid.uuid4()),
+                plan_id=plan_id,
+                sequence=0,
+                operation="move_folder",
+                source_path=str(folder),
+                destination_path=str(lib / "2019 Rome"),
+                journal_state="moving",
+            )
+        )
+        session.commit()
+
+    report = _service(sf, config).preflight(location["id"])
+
+    assert report["state"] == "blocked" and "rename_pending" in _codes(report)
+
+
+# ── token ────────────────────────────────────────────────────────────────────
+
+
+def test_token_is_stable_while_nothing_relevant_changes(tmp_path):
+    config, sf, lib, archive = _env(tmp_path)
+    _album(sf, lib)
+    location = _location(sf, config, archive)
+    service = _service(sf, config)
+
+    first = service.preflight(location["id"])["token"]
+
+    assert service.preflight(location["id"])["token"] == first
+    assert service.verify_token(first, location["id"]) is True
+
+
+def test_an_edited_source_makes_the_token_stale(tmp_path):
+    config, sf, lib, archive = _env(tmp_path)
+    folder, _ = _album(sf, lib)
+    location = _location(sf, config, archive)
+    service = _service(sf, config)
+    token = service.preflight(location["id"])["token"]
+
+    (folder / "b.jpg").write_bytes(b"edited outside the app")
+
+    assert service.verify_token(token, location["id"]) is False
+
+
+def test_a_changed_destination_makes_the_token_stale(tmp_path):
+    config, sf, lib, archive = _env(tmp_path)
+    _album(sf, lib)
+    location = _location(sf, config, archive)
+    service = _service(sf, config)
+    token = service.preflight(location["id"])["token"]
+
+    (archive / "rome").mkdir()
+    (archive / "rome" / "a.jpg").write_bytes(b"appeared after approval")
+
+    assert service.verify_token(token, location["id"]) is False
+
+
+def test_a_token_from_another_scope_or_medium_is_rejected(tmp_path):
+    config, sf, lib, archive = _env(tmp_path)
+    _album(sf, lib, "rome")
+    _album(sf, lib, "paris", names=("c.jpg",))
+    other = tmp_path / "archive2"
+    other.mkdir()
+    location = _location(sf, config, archive)
+    second = _location(sf, config, other, name="second")
+    service = _service(sf, config)
+
+    rome = service.preflight(location["id"], ["rome"])["token"]
+
+    assert service.verify_token(rome, location["id"], ["paris"]) is False
+    assert service.verify_token(rome, second["id"], ["rome"]) is False
+    assert service.verify_token("v1:not-a-real-token", location["id"]) is False
+    assert service.verify_token("", location["id"]) is False
+
+
+def test_the_token_survives_free_space_and_timestamp_drift(tmp_path):
+    """Free space changes constantly on a live disk; a token that expired on every
+    byte written elsewhere would train users to ignore it."""
+    config, sf, lib, archive = _env(tmp_path)
+    _album(sf, lib)
+    location = _location(sf, config, archive)
+    service = _service(sf, config)
+    token = service.preflight(location["id"])["token"]
+
+    (tmp_path / "unrelated.bin").write_bytes(b"0" * 100_000)
+    with sf() as session:
+        from photo_pipeline.models import ArchiveLocation
+
+        session.get(ArchiveLocation, location["id"]).last_seen_at = NOW - timedelta(days=5)
+        session.commit()
+
+    assert service.verify_token(token, location["id"]) is True
+
+
+# ── API surface ──────────────────────────────────────────────────────────────
+
+
+def test_api_registers_a_location_and_returns_a_preflight_report(tmp_path):
+    config, sf, lib, archive = _env(tmp_path)
+    _album(sf, lib)
+
+    with TestClient(create_app(config)) as client:
+        created = client.post(
+            "/api/v1/archive-locations", json={"name": "external", "root": str(archive)}
+        )
+        listed = client.get("/api/v1/archive-locations")
+        report = client.post(
+            "/api/v1/archive-preflight", json={"location_id": created.json()["id"]}
+        )
+
+    assert created.status_code == 201
+    assert [row["name"] for row in listed.json()["locations"]] == ["external"]
+    assert report.status_code == 200
+    assert report.json()["state"] == "ready" and report.json()["token"].startswith("v1:")
+
+
+def test_api_rejects_an_unknown_location_and_an_unsafe_root(tmp_path):
+    config, sf, lib, _ = _env(tmp_path)
+
+    with TestClient(create_app(config)) as client:
+        unknown = client.post("/api/v1/archive-preflight", json={"location_id": "nope"})
+        unsafe = client.post("/api/v1/archive-locations", json={"name": "bad", "root": str(lib)})
+
+    assert unknown.status_code == 404 and unknown.json()["error"]["code"] == "unknown_location"
+    assert unsafe.status_code == 422 and unsafe.json()["error"]["code"] == "unsafe_destination"

tests/integration/test_archive_recovery.py

@@ -0,0 +1,313 @@
+"""Recovering interrupted archive transfers (US06-02).
+
+The crash tests are real: a child process applies an archive plan and is killed by
+the ``PHOTO_PIPELINE_FAULT_AFTER`` barrier at each persisted transition in turn —
+including the moment immediately after an active source has been unlinked. The
+parent then reopens the database and asserts the one invariant archiving exists to
+uphold: **no verified file is ever lost, and no source is removed without a durable,
+byte-identical archive copy.**
+
+Both transfer paths are exercised: the same-filesystem atomic move and (with the
+child forcing the device comparison) the cross-filesystem copy/verify/publish.
+"""
+
+import os
+import subprocess
+import sys
+import uuid
+from datetime import datetime, timezone
+from pathlib import Path
+
+import pytest
+from sqlalchemy import select
+
+from photo_pipeline.config import Config
+from photo_pipeline.db import create_db_engine, create_session_factory, run_migrations
+from photo_pipeline.models import Asset, UploadBatch, UploadItem
+from photo_pipeline.services.archive_journal import FORWARD, MANUAL, RESUMABLE, ArchiveState
+from photo_pipeline.services.archive_transfer import (
+    MANIFEST_NAME,
+    ArchiveTransferService,
+    read_manifest,
+)
+from photo_pipeline.services.archives import ArchiveService
+from photo_pipeline.services.hashing import sha256_file
+
+pytestmark = pytest.mark.phase_f
+
+REPO = Path(__file__).resolve().parents[2]
+NOW = datetime(2026, 1, 1, tzinfo=timezone.utc)
+
+# Barriers, in the order the transfer persists them. ``source_removed`` is the one
+# that matters most: the original is already gone at that point.
+BARRIERS = [
+    ArchiveState.TRANSFERRING,
+    ArchiveState.VERIFIED,
+    ArchiveState.REMOVING,
+    "source_removed",
+    ArchiveState.COMPLETE,
+]
+
+# A child that applies the plan and dies at the configured barrier. ``force_copy``
+# makes it take the cross-filesystem path without a second real volume.
+APPLY_SCRIPT = """
+import sys
+sys.path.insert(0, {repo!r})
+from photo_pipeline.config import Config
+from photo_pipeline.db import create_db_engine, create_session_factory
+from photo_pipeline.services import archive_transfer
+
+db_url, data_dir, lib, plan_id, force_copy = sys.argv[1:6]
+if force_copy == "1":
+    archive_transfer._same_filesystem = lambda *args: False
+config = Config.from_env(
+    {{"PHOTO_PIPELINE_DATA_DIR": data_dir, "PHOTO_PIPELINE_LIBRARY_ROOTS": lib}}
+)
+sf = create_session_factory(create_db_engine(db_url))
+archive_transfer.ArchiveTransferService(sf, config=config).apply(plan_id)
+"""
+
+
+# ── environment ──────────────────────────────────────────────────────────────
+
+
+def _env(tmp_path):
+    (tmp_path / "data").mkdir(exist_ok=True)
+    lib = tmp_path / "lib"
+    lib.mkdir(exist_ok=True)
+    archive = tmp_path / "archive"
+    archive.mkdir(exist_ok=True)
+    config = Config.from_env(
+        {
+            "PHOTO_PIPELINE_DATA_DIR": str(tmp_path / "data"),
+            "PHOTO_PIPELINE_LIBRARY_ROOTS": str(lib),
+            "PHOTO_PIPELINE_ARCHIVE_FREE_SPACE_RESERVE_BYTES": "0",
+        }
+    )
+    run_migrations(config.database_url)
+    return config, create_session_factory(create_db_engine(config.database_url)), lib, archive
+
+
+def _album(sf, lib, album="rome", names=("a.jpg", "b.jpg")):
+    folder = lib / album
+    folder.mkdir(parents=True, exist_ok=True)
+    with sf() as session:
+        batch_id = str(uuid.uuid4())
+        session.add(
+            UploadBatch(
+                id=batch_id,
+                album=album,
+                folder=str(folder),
+                album_name=album,
+                state="succeeded",
+                preflight_token="v1:test",
+                outcome_state="verified",
+                created_at=NOW,
+            )
+        )
+        for name in names:
+            path = folder / name
+            path.write_bytes(f"{album}/{name} content".encode() * 8)
+            asset_id = str(uuid.uuid4())
+            session.add(
+                Asset(
+                    id=asset_id,
+                    original_path=str(path),
+                    current_path=str(path),
+                    discovered_at=NOW,
+                    hash_version=1,
+                    byte_size=path.stat().st_size,
+                    current_sha256=sha256_file(path),
+                )
+            )
+            session.add(
+                UploadItem(
+                    batch_id=batch_id,
+                    asset_id=asset_id,
+                    path=str(path),
+                    sha256=sha256_file(path),
+                    sha1="0" * 40,
+                    state="sent",
+                    outcome="uploaded",
+                )
+            )
+        session.commit()
+    return folder
+
+
+def _plan(sf, config, archive, albums=None):
+    location = ArchiveService(sf, config=config).register("external", str(archive))
+    token = ArchiveService(sf, config=config).preflight(location["id"], albums)["token"]
+    service = ArchiveTransferService(sf, config=config)
+    return service, service.create(location["id"], albums, token=token)
+
+
+def _crash_during_apply(config, tmp_path, lib, plan, barrier, *, force_copy=False):
+    script = tmp_path / f"apply_{barrier}.py"
+    script.write_text(APPLY_SCRIPT.format(repo=str(REPO)))
+    env = dict(os.environ)
+    env["PHOTO_PIPELINE_FAULT_AFTER"] = barrier
+    result = subprocess.run(
+        [
+            sys.executable,
+            str(script),
+            config.database_url,
+            str(tmp_path / "data"),
+            str(lib),
+            plan["id"],
+            "1" if force_copy else "0",
+        ],
+        env=env,
+        capture_output=True,
+    )
+    assert result.returncode in (9, -9), (
+        f"child should have been killed at {barrier}, got {result.returncode}: "
+        f"{result.stderr.decode(errors='replace')[-400:]}"
+    )
+
+
+def _contents(*roots):
+    return sorted(
+        path.read_bytes()
+        for root in roots
+        for path in root.rglob("*")
+        if path.is_file() and path.name != MANIFEST_NAME and not path.name.startswith(".")
+    )
+
+
+def _reopen(config):
+    return create_session_factory(create_db_engine(config.database_url))
+
+
+# ── fault injection at every persisted transition ────────────────────────────
+
+
+@pytest.mark.parametrize("barrier", BARRIERS)
+@pytest.mark.parametrize("force_copy", [False, True], ids=["move", "copy"])
+def test_a_crash_at_every_transition_loses_nothing_and_recovers(tmp_path, barrier, force_copy):
+    config, sf, lib, archive = _env(tmp_path)
+    _album(sf, lib)
+    service, plan = _plan(sf, config, archive)
+    before = _contents(lib)
+
+    _crash_during_apply(config, tmp_path, lib, plan, barrier, force_copy=force_copy)
+
+    # Every file still exists somewhere: the crash may not have cost a single byte.
+    assert set(before) <= set(_contents(lib, archive)), f"content lost at {barrier}"
+
+    reopened = _reopen(config)
+    recovery = ArchiveTransferService(reopened, config=config)
+    # Whatever the crash interrupted is recoverable from evidence — never ambiguous.
+    # (A crash on the last barrier can land on a terminal item and leave none.)
+    verdicts = recovery.journal.classify_all()
+    assert all(v["classification"] in {RESUMABLE, FORWARD} for v in verdicts), verdicts
+
+    recovery.recover()
+    recovery.apply(plan["id"])  # finish whatever the crash never started
+
+    assert _contents(archive) == before
+    assert _contents(lib) == []
+    assert recovery.journal.blocks_mutation() is False
+    assert recovery.journal.plan_state(plan["id"]) == "complete"
+
+
+@pytest.mark.parametrize("barrier", [ArchiveState.VERIFIED, "source_removed"])
+def test_recovery_is_idempotent_across_repeated_restarts(tmp_path, barrier):
+    config, sf, lib, archive = _env(tmp_path)
+    _album(sf, lib)
+    service, plan = _plan(sf, config, archive)
+    _crash_during_apply(config, tmp_path, lib, plan, barrier)
+
+    reopened = _reopen(config)
+    recovery = ArchiveTransferService(reopened, config=config)
+    recovery.recover()
+    settled = (_contents(lib), _contents(archive), read_manifest(archive / "rome"))
+
+    for _ in range(2):
+        recovery.recover()
+        assert (_contents(lib), _contents(archive), read_manifest(archive / "rome")) == settled
+
+
+def test_a_crash_before_anything_was_published_leaves_the_source_intact(tmp_path):
+    config, sf, lib, archive = _env(tmp_path)
+    folder = _album(sf, lib, names=("a.jpg",))
+    service, plan = _plan(sf, config, archive)
+
+    _crash_during_apply(config, tmp_path, lib, plan, ArchiveState.TRANSFERRING)
+
+    recovery = ArchiveTransferService(_reopen(config), config=config)
+    verdict = recovery.journal.classify_all()[0]
+    assert verdict["classification"] == RESUMABLE
+    assert (folder / "a.jpg").exists()
+    assert not (archive / "rome" / "a.jpg").exists()
+
+    assert recovery.recover() == {"resumed": 1, "completed": 0, "manual": 0}
+    assert recovery.journal.operations(plan["id"])[0]["journal_state"] == ArchiveState.PLANNED
+
+
+def test_a_crash_after_the_source_was_removed_finishes_the_bookkeeping(tmp_path):
+    """The dangerous window: the original is gone and the database still points at
+    it. Recovery must complete the record, never re-transfer or report loss."""
+    config, sf, lib, archive = _env(tmp_path)
+    folder = _album(sf, lib, names=("a.jpg",))
+    service, plan = _plan(sf, config, archive)
+    expected = service.journal.operations(plan["id"])[0]["expected_sha256"]
+
+    _crash_during_apply(config, tmp_path, lib, plan, "source_removed")
+
+    reopened = _reopen(config)
+    recovery = ArchiveTransferService(reopened, config=config)
+    verdict = recovery.journal.classify_all()[0]
+    assert verdict["classification"] == FORWARD
+    assert not (folder / "a.jpg").exists()
+    assert sha256_file(archive / "rome" / "a.jpg") == expected
+    with reopened() as session:
+        stranded = session.scalar(select(Asset))
+        assert stranded.current_path is not None, "the crash happened before the DB update"
+
+    assert recovery.recover() == {"resumed": 0, "completed": 1, "manual": 0}
+
+    with reopened() as session:
+        asset = session.scalar(select(Asset))
+        assert asset.current_path is None
+        assert asset.availability_state == "archived_online"
+        assert asset.archive_path == "rome/a.jpg"
+    assert recovery.journal.operations(plan["id"])[0]["journal_state"] == ArchiveState.COMPLETE
+    assert len(read_manifest(archive / "rome")) == 1, "the manifest is not duplicated"
+
+
+def test_an_archive_copy_that_changed_after_the_crash_blocks_for_a_human(tmp_path):
+    """Wrong bytes at the destination can never justify deleting the original, and
+    are never overwritten either."""
+    config, sf, lib, archive = _env(tmp_path)
+    folder = _album(sf, lib, names=("a.jpg",))
+    service, plan = _plan(sf, config, archive)
+    _crash_during_apply(config, tmp_path, lib, plan, ArchiveState.VERIFIED, force_copy=True)
+    (archive / "rome" / "a.jpg").write_bytes(b"tampered with while the app was down")
+
+    recovery = ArchiveTransferService(_reopen(config), config=config)
+    verdict = recovery.journal.classify_all()[0]
+
+    assert verdict["classification"] == MANUAL
+    assert recovery.recover() == {"resumed": 0, "completed": 0, "manual": 1}
+    assert (folder / "a.jpg").exists(), "the source is kept while the archive is unproven"
+    assert (archive / "rome" / "a.jpg").read_bytes() == b"tampered with while the app was down"
+    # And the unresolved item keeps blocking further archiving.
+    assert recovery.journal.blocks_mutation() is True
+    report = ArchiveService(_reopen(config), config=config).preflight(plan["location_id"])
+    assert "archive_pending" in {issue["code"] for issue in report["blockers"]}
+
+
+def test_a_verified_archive_whose_copy_vanished_is_never_reported_as_archived(tmp_path):
+    config, sf, lib, archive = _env(tmp_path)
+    folder = _album(sf, lib, names=("a.jpg",))
+    service, plan = _plan(sf, config, archive)
+    _crash_during_apply(config, tmp_path, lib, plan, ArchiveState.REMOVING, force_copy=True)
+    (archive / "rome" / "a.jpg").unlink()  # the medium lost it
+
+    recovery = ArchiveTransferService(_reopen(config), config=config)
+
+    assert recovery.journal.classify_all()[0]["classification"] == MANUAL
+    assert recovery.recover()["manual"] == 1
+    assert (folder / "a.jpg").exists()
+    assert recovery.journal.operations(plan["id"])[0]["journal_state"] != ArchiveState.COMPLETE

tests/integration/test_archive_transfer.py

@@ -0,0 +1,480 @@
+"""Transferring, verifying, and removing active sources (US06-02).
+
+Every case here asks the same question the service exists to answer: could an
+original leave active storage without a durable, byte-identical archive copy? The
+files are real, the hashes are real, and each failure path asserts that the source
+is still exactly where it was.
+
+Cross-filesystem behaviour is forced by patching the device comparison rather than
+by requiring a second real filesystem in CI — the copy/verify/publish code that runs
+is the production one.
+"""
+
+import json
+import uuid
+from datetime import datetime, timezone
+
+import pytest
+from fastapi.testclient import TestClient
+from sqlalchemy import select
+
+from photo_pipeline.api.app import create_app
+from photo_pipeline.config import Config
+from photo_pipeline.db import create_db_engine, create_session_factory, run_migrations
+from photo_pipeline.models import Asset, AssetPath, UploadBatch, UploadItem
+from photo_pipeline.services import archive_transfer as transfer_module
+from photo_pipeline.services.archive_journal import ArchiveState
+from photo_pipeline.services.archive_transfer import (
+    MANIFEST_NAME,
+    ArchiveTransferService,
+    read_manifest,
+)
+from photo_pipeline.services.archives import ArchiveError, ArchiveService
+from photo_pipeline.services.hashing import sha256_file
+
+pytestmark = pytest.mark.phase_f  # part of the Phase F acceptance gate (US06-06)
+
+NOW = datetime(2026, 1, 1, tzinfo=timezone.utc)
+
+
+# ── environment ──────────────────────────────────────────────────────────────
+
+
+def _env(tmp_path):
+    (tmp_path / "data").mkdir(exist_ok=True)
+    lib = tmp_path / "lib"
+    lib.mkdir(exist_ok=True)
+    archive = tmp_path / "archive"
+    archive.mkdir(exist_ok=True)
+    config = Config.from_env(
+        {
+            "PHOTO_PIPELINE_DATA_DIR": str(tmp_path / "data"),
+            "PHOTO_PIPELINE_LIBRARY_ROOTS": str(lib),
+            "PHOTO_PIPELINE_ARCHIVE_FREE_SPACE_RESERVE_BYTES": "0",
+        }
+    )
+    run_migrations(config.database_url)
+    return config, create_session_factory(create_db_engine(config.database_url)), lib, archive
+
+
+def _album(sf, lib, album="rome", names=("a.jpg", "b.jpg")):
+    """A real album whose assets carry the verified upload evidence archiving needs."""
+    folder = lib / album
+    folder.mkdir(parents=True, exist_ok=True)
+    ids = []
+    with sf() as session:
+        batch_id = str(uuid.uuid4())
+        session.add(
+            UploadBatch(
+                id=batch_id,
+                album=album,
+                folder=str(folder),
+                album_name=album,
+                state="succeeded",
+                preflight_token="v1:test",
+                outcome_state="verified",
+                created_at=NOW,
+            )
+        )
+        for name in names:
+            path = folder / name
+            path.write_bytes(f"{album}/{name} bytes".encode() * 8)
+            asset_id = str(uuid.uuid4())
+            ids.append(asset_id)
+            session.add(
+                Asset(
+                    id=asset_id,
+                    original_path=str(path),
+                    current_path=str(path),
+                    discovered_at=NOW,
+                    hash_version=1,
+                    byte_size=path.stat().st_size,
+                    current_sha256=sha256_file(path),
+                )
+            )
+            session.add(AssetPath(asset_id=asset_id, path=str(path), valid_from=NOW))
+            session.add(
+                UploadItem(
+                    batch_id=batch_id,
+                    asset_id=asset_id,
+                    path=str(path),
+                    sha256=sha256_file(path),
+                    sha1="0" * 40,
+                    state="sent",
+                    outcome="uploaded",
+                )
+            )
+        session.commit()
+    return folder, ids
+
+
+def _location(sf, config, archive, name="external"):
+    return ArchiveService(sf, config=config).register(name, str(archive))
+
+
+def _plan(sf, config, location_id, albums=None):
+    service = ArchiveTransferService(sf, config=config)
+    token = ArchiveService(sf, config=config).preflight(location_id, albums)["token"]
+    return service, service.create(location_id, albums, token=token)
+
+
+def _contents(*roots):
+    """Every file's bytes under the given roots, ignoring our own bookkeeping."""
+    return sorted(
+        path.read_bytes()
+        for root in roots
+        for path in root.rglob("*")
+        if path.is_file() and path.name != MANIFEST_NAME and not path.name.startswith(".")
+    )
+
+
+def _states(service, plan_id):
+    return [row["journal_state"] for row in service.journal.operations(plan_id)]
+
+
+def _asset(sf, asset_id):
+    with sf() as session:
+        return session.get(Asset, asset_id)
+
+
+# ── planning ─────────────────────────────────────────────────────────────────
+
+
+def test_a_plan_records_every_file_with_its_expected_hash_and_moves_nothing(tmp_path):
+    config, sf, lib, archive = _env(tmp_path)
+    folder, ids = _album(sf, lib)
+    location = _location(sf, config, archive)
+    before = _contents(lib)
+
+    service, plan = _plan(sf, config, location["id"])
+
+    assert plan["state"] == "planned" and plan["asset_count"] == 2
+    assert sorted(op["asset_id"] for op in plan["operations"]) == sorted(ids)
+    for operation in plan["operations"]:
+        source = folder / operation["archive_path"].split("/")[-1]
+        assert operation["expected_sha256"] == sha256_file(source)
+        assert operation["archive_path"].startswith("rome/")
+        assert operation["journal_state"] == ArchiveState.PLANNED
+    assert _contents(lib) == before
+    assert _contents(archive) == []
+
+
+def test_a_stale_token_cannot_create_a_plan(tmp_path):
+    config, sf, lib, archive = _env(tmp_path)
+    folder, _ = _album(sf, lib)
+    location = _location(sf, config, archive)
+    token = ArchiveService(sf, config=config).preflight(location["id"])["token"]
+
+    (folder / "a.jpg").write_bytes(b"edited after approval")
+
+    with pytest.raises(ArchiveError) as error:
+        ArchiveTransferService(sf, config=config).create(location["id"], token=token)
+    assert error.value.code == "stale_token"
+
+
+def test_a_blocked_scope_cannot_create_a_plan(tmp_path):
+    """No verified upload means Immich may not hold these bytes; archiving would
+    remove the only copy."""
+    config, sf, lib, archive = _env(tmp_path)
+    folder = lib / "rome"
+    folder.mkdir()
+    path = folder / "a.jpg"
+    path.write_bytes(b"never uploaded")
+    with sf() as session:
+        session.add(
+            Asset(
+                id=str(uuid.uuid4()),
+                original_path=str(path),
+                current_path=str(path),
+                discovered_at=NOW,
+                hash_version=1,
+                byte_size=path.stat().st_size,
+                current_sha256=sha256_file(path),
+            )
+        )
+        session.commit()
+    location = _location(sf, config, archive)
+    token = ArchiveService(sf, config=config).preflight(location["id"])["token"]
+
+    with pytest.raises(ArchiveError) as error:
+        ArchiveTransferService(sf, config=config).create(location["id"], token=token)
+
+    assert error.value.code == "blocked"
+    assert path.exists()
+
+
+# ── the transfer ─────────────────────────────────────────────────────────────
+
+
+@pytest.mark.parametrize("same_filesystem", [True, False])
+def test_archiving_verifies_the_copy_before_the_source_is_removed(
+    tmp_path, monkeypatch, same_filesystem
+):
+    config, sf, lib, archive = _env(tmp_path)
+    folder, ids = _album(sf, lib)
+    location = _location(sf, config, archive)
+    service, plan = _plan(sf, config, location["id"])
+    expected = {op["asset_id"]: op["expected_sha256"] for op in plan["operations"]}
+    before = _contents(lib)
+    if not same_filesystem:
+        # Force the copy-verify-publish path without needing a second real volume.
+        monkeypatch.setattr(transfer_module, "_same_filesystem", lambda *_: False)
+
+    result = service.apply(plan["id"])
+
+    assert result == {
+        "plan_id": plan["id"],
+        "archived": 2,
+        "failed": 0,
+        "skipped": 0,
+        "state": "complete",
+    }
+    # The bytes moved: nothing is left in the library, everything is in the archive.
+    assert _contents(archive) == before
+    assert _contents(lib) == []
+    assert not folder.exists(), "an emptied album folder is not left behind"
+    for asset_id, digest in expected.items():
+        archived = archive / "rome" / _asset(sf, asset_id).archive_path.split("/")[-1]
+        assert sha256_file(archived) == digest
+    assert _states(service, plan["id"]) == [ArchiveState.COMPLETE] * 2
+    # No transfer temporaries survive either path.
+    assert not list((archive / "rome").glob(".archive-*"))
+
+
+@pytest.mark.parametrize("same_filesystem", [True, False])
+def test_the_manifest_on_the_medium_matches_the_archived_bytes(
+    tmp_path, monkeypatch, same_filesystem
+):
+    """The manifest is the medium's own record: it must be usable to verify the
+    archive with no database at all."""
+    config, sf, lib, archive = _env(tmp_path)
+    _album(sf, lib)
+    location = _location(sf, config, archive)
+    service, plan = _plan(sf, config, location["id"])
+    if not same_filesystem:
+        monkeypatch.setattr(transfer_module, "_same_filesystem", lambda *_: False)
+
+    service.apply(plan["id"])
+
+    entries = read_manifest(archive / "rome")
+    assert len(entries) == 2
+    for entry in entries:
+        archived = archive / entry["archive_path"]
+        assert sha256_file(archived) == entry["sha256"]
+        assert entry["byte_size"] == archived.stat().st_size
+        assert entry["media_id"] == location["media_id"]
+        assert entry["plan_id"] == plan["id"] and entry["album"] == "rome"
+
+
+def test_archived_assets_keep_their_identity_and_gain_their_new_location(tmp_path):
+    config, sf, lib, archive = _env(tmp_path)
+    folder, ids = _album(sf, lib)
+    location = _location(sf, config, archive)
+    service, plan = _plan(sf, config, location["id"])
+
+    service.apply(plan["id"])
+
+    with sf() as session:
+        for asset_id in ids:
+            asset = session.get(Asset, asset_id)
+            assert asset is not None, "archiving never deletes the record"
+            assert asset.current_path is None
+            assert asset.availability_state == "archived_online"
+            assert asset.archive_location_id == location["id"]
+            assert (archive / asset.archive_path).exists()
+            occurrences = session.scalars(
+                select(AssetPath).where(AssetPath.asset_id == asset_id)
+            ).all()
+            active = [row for row in occurrences if row.valid_until is None]
+            assert [row.path for row in active] == [str(archive / asset.archive_path)]
+            closed = [row for row in occurrences if row.valid_until is not None]
+            assert [row.path for row in closed] == [str(folder / asset.archive_path.split("/")[-1])]
+
+
+def test_applying_the_same_plan_again_changes_nothing(tmp_path):
+    config, sf, lib, archive = _env(tmp_path)
+    _album(sf, lib)
+    location = _location(sf, config, archive)
+    service, plan = _plan(sf, config, location["id"])
+    service.apply(plan["id"])
+    archived = _contents(archive)
+    manifest = read_manifest(archive / "rome")
+
+    again = service.apply(plan["id"])
+
+    assert again["skipped"] == 2 and again["archived"] == 0 and again["failed"] == 0
+    assert _contents(archive) == archived
+    assert read_manifest(archive / "rome") == manifest
+
+
+# ── unexpected changes stop the item ─────────────────────────────────────────
+
+
+def test_an_occupied_destination_is_never_overwritten(tmp_path):
+    config, sf, lib, archive = _env(tmp_path)
+    folder, _ = _album(sf, lib, names=("a.jpg",))
+    location = _location(sf, config, archive)
+    service, plan = _plan(sf, config, location["id"])
+    (archive / "rome").mkdir()
+    (archive / "rome" / "a.jpg").write_bytes(b"someone else's file")
+
+    result = service.apply(plan["id"])
+
+    assert result["failed"] == 1 and result["state"] == "failed"
+    assert (archive / "rome" / "a.jpg").read_bytes() == b"someone else's file"
+    assert (folder / "a.jpg").exists(), "the source must survive a refused transfer"
+    assert service.journal.operations(plan["id"])[0]["error_code"] == "destination_exists"
+
+
+def test_a_source_edited_after_planning_is_neither_archived_nor_removed(tmp_path):
+    config, sf, lib, archive = _env(tmp_path)
+    folder, _ = _album(sf, lib, names=("a.jpg",))
+    location = _location(sf, config, archive)
+    service, plan = _plan(sf, config, location["id"])
+
+    (folder / "a.jpg").write_bytes(b"edited between approval and apply")
+    result = service.apply(plan["id"])
+
+    assert result["failed"] == 1
+    assert (folder / "a.jpg").read_bytes() == b"edited between approval and apply"
+    assert not (archive / "rome" / "a.jpg").exists()
+    assert service.journal.operations(plan["id"])[0]["error_code"] == "source_changed"
+
+
+def test_a_copy_that_lands_with_the_wrong_bytes_is_not_published(tmp_path, monkeypatch):
+    """The read-back hash — not the fact that a write returned — is what proves the
+    archive copy."""
+    config, sf, lib, archive = _env(tmp_path)
+    folder, _ = _album(sf, lib, names=("a.jpg",))
+    location = _location(sf, config, archive)
+    service, plan = _plan(sf, config, location["id"])
+    monkeypatch.setattr(transfer_module, "_same_filesystem", lambda *_: False)
+    monkeypatch.setattr(
+        transfer_module.shutil,
+        "copyfileobj",
+        lambda src, dst, length=0: dst.write(b"corrupted in flight"),
+    )
+
+    result = service.apply(plan["id"])
+
+    assert result["failed"] == 1
+    assert service.journal.operations(plan["id"])[0]["error_code"] == "copy_mismatch"
+    assert (folder / "a.jpg").exists()
+    assert not (archive / "rome" / "a.jpg").exists()
+    assert not list((archive / "rome").glob(".archive-*")), "the failed copy is cleaned up"
+
+
+def test_a_source_replaced_by_a_symlink_is_refused(tmp_path):
+    config, sf, lib, archive = _env(tmp_path)
+    folder, _ = _album(sf, lib, names=("a.jpg",))
+    location = _location(sf, config, archive)
+    service, plan = _plan(sf, config, location["id"])
+    elsewhere = tmp_path / "elsewhere.jpg"
+    elsewhere.write_bytes(b"not a library file")
+    (folder / "a.jpg").unlink()
+    (folder / "a.jpg").symlink_to(elsewhere)
+
+    result = service.apply(plan["id"])
+
+    assert result["failed"] == 1
+    assert elsewhere.exists() and (folder / "a.jpg").is_symlink()
+    assert not (archive / "rome" / "a.jpg").exists()
+
+
+def test_one_failed_file_does_not_stop_the_rest_of_the_album(tmp_path):
+    config, sf, lib, archive = _env(tmp_path)
+    folder, _ = _album(sf, lib, names=("a.jpg", "b.jpg"))
+    location = _location(sf, config, archive)
+    service, plan = _plan(sf, config, location["id"])
+    (folder / "a.jpg").write_bytes(b"edited after approval")
+
+    result = service.apply(plan["id"])
+
+    assert (result["archived"], result["failed"]) == (1, 1)
+    assert (folder / "a.jpg").exists() and not (folder / "b.jpg").exists()
+    assert (archive / "rome" / "b.jpg").exists()
+    assert folder.exists(), "a folder that still holds an unarchived file stays"
+
+
+def test_an_unresolved_transfer_blocks_the_next_preflight(tmp_path, monkeypatch):
+    config, sf, lib, archive = _env(tmp_path)
+    _album(sf, lib, "rome", names=("a.jpg",))
+    _album(sf, lib, "paris", names=("c.jpg",))
+    location = _location(sf, config, archive)
+    service, plan = _plan(sf, config, location["id"], ["rome"])
+    # Leave one item stuck mid-transfer, as a killed process would.
+    service.journal.begin(
+        service.journal.operations(plan["id"])[0]["id"], worker_id="w", fencing_token=1
+    )
+
+    report = ArchiveService(sf, config=config).preflight(location["id"], ["paris"])
+
+    assert report["state"] == "blocked"
+    assert "archive_pending" in {issue["code"] for issue in report["blockers"]}
+
+
+# ── API surface ──────────────────────────────────────────────────────────────
+
+
+def test_api_creates_a_plan_and_queues_it_on_the_archiver_lane(tmp_path):
+    config, sf, lib, archive = _env(tmp_path)
+    _album(sf, lib)
+
+    with TestClient(create_app(config)) as client:
+        location = client.post(
+            "/api/v1/archive-locations", json={"name": "external", "root": str(archive)}
+        ).json()
+        token = client.post(
+            "/api/v1/archive-preflight", json={"location_id": location["id"]}
+        ).json()["token"]
+        created = client.post(
+            "/api/v1/archive-plans", json={"location_id": location["id"], "token": token}
+        )
+        plan_id = created.json()["id"]
+        fetched = client.get(f"/api/v1/archive-plans/{plan_id}")
+        applied = client.post(f"/api/v1/archive-plans/{plan_id}/apply")
+        listed = client.get("/api/v1/archive-plans")
+
+    assert created.status_code == 201 and created.json()["asset_count"] == 2
+    assert fetched.status_code == 200 and len(fetched.json()["operations"]) == 2
+    assert applied.status_code == 200 and applied.json()["job"]["state"] == "queued"
+    assert applied.json()["job"]["lock_key"] == "archive"
+    assert [row["id"] for row in listed.json()["plans"]] == [plan_id]
+    # Queuing alone must not have touched a single file.
+    assert _contents(lib) and _contents(archive) == []
+
+
+def test_api_refuses_a_stale_token_and_an_unknown_plan(tmp_path):
+    config, sf, lib, archive = _env(tmp_path)
+    _album(sf, lib)
+    location = _location(sf, config, archive)
+
+    with TestClient(create_app(config)) as client:
+        stale = client.post(
+            "/api/v1/archive-plans",
+            json={"location_id": location["id"], "token": "v1:not-the-real-token"},
+        )
+        unknown = client.get("/api/v1/archive-plans/nope")
+        unknown_apply = client.post("/api/v1/archive-plans/nope/apply")
+
+    assert stale.status_code == 409 and stale.json()["error"]["code"] == "stale_token"
+    assert unknown.status_code == 404 and unknown_apply.status_code == 404
+
+
+def test_api_reports_recovery_state_for_an_interrupted_transfer(tmp_path):
+    config, sf, lib, archive = _env(tmp_path)
+    _album(sf, lib, names=("a.jpg",))
+    location = _location(sf, config, archive)
+    service, plan = _plan(sf, config, location["id"])
+    service.journal.begin(
+        service.journal.operations(plan["id"])[0]["id"], worker_id="w", fencing_token=1
+    )
+
+    with TestClient(create_app(config)) as client:
+        status = client.get("/api/v1/archive-recovery").json()
+        resolved = client.post("/api/v1/archive-recovery/resolve").json()
+
+    assert status["blocks_mutation"] is True
+    assert status["operations"][0]["classification"] == "resumable"
+    assert resolved == {"resumed": 1, "completed": 0, "manual": 0}
+    assert json.loads(json.dumps(resolved))  # plain JSON, nothing exotic

tests/integration/test_inventory_reconcile.py

@@ -167,7 +167,8 @@ def test_missing_file_is_flagged_not_deleted(
     assets = assets_by_id(make_factory(db_url))
     assert missing_id in assets  # not pruned
     assert assets[missing_id].missing_at is not None
-    assert assets[missing_id].availability_state == "active"
+    # Nothing explains the absence: this is not an offline archive medium (US06-03).
+    assert assets[missing_id].availability_state == "missing_unexpected"
 
 
 def test_reappearing_file_clears_missing(
@@ -186,6 +187,7 @@ def test_reappearing_file_clears_missing(
     inventory.scan(lib)
     assets = assets_by_id(make_factory(db_url))
     assert assets[asset_id].missing_at is None
+    assert assets[asset_id].availability_state == "active"
 
 
 def test_identity_and_state_durable_across_restart(

tests/integration/test_offline_assets.py

@@ -0,0 +1,377 @@
+"""Offline identity and review evidence (US06-03).
+
+An archived photo is not gone: it keeps its identity, its hashes, and enough
+evidence to be recognised in a duplicate cluster while its medium sits in a
+drawer. Every case here archives a *real* album through the real transfer, then
+takes the medium away by removing its marker — the same thing the service sees
+when an external disk is unplugged — and asks whether the application still tells
+the truth about where the bytes are.
+
+The distinction that matters throughout: an unmounted medium is
+``archived_offline`` (expected, harmless), a mounted medium with a hole in it is
+``missing_unexpected`` (needs a human). Confusing the two is how an archive
+quietly loses a photo.
+"""
+
+import shutil
+import uuid
+from datetime import datetime, timezone
+
+import numpy as np
+import pytest
+from fastapi.testclient import TestClient
+from PIL import Image
+from sqlalchemy import select
+
+from photo_pipeline.api.app import create_app
+from photo_pipeline.config import Config
+from photo_pipeline.db import create_db_engine, create_session_factory, run_migrations
+from photo_pipeline.models import Asset, Thumbnail, UploadBatch, UploadItem
+from photo_pipeline.services.archive_transfer import ArchiveTransferService
+from photo_pipeline.services.archives import MARKER_NAME, ArchiveService
+from photo_pipeline.services.availability import (
+    ACTIVE,
+    ARCHIVED_OFFLINE,
+    ARCHIVED_ONLINE,
+    MISSING_UNEXPECTED,
+)
+from photo_pipeline.services.duplicates import ClusterState, DuplicateService, Method
+from photo_pipeline.services.hashing import sha256_file
+from photo_pipeline.services.inventory import InventoryService
+from photo_pipeline.services.thumbnails import PROTECTED_SIZE, ThumbnailService, ThumbnailUnavailable
+
+pytestmark = pytest.mark.phase_f  # part of the Phase F acceptance gate (US06-06)
+
+NOW = datetime(2026, 1, 1, tzinfo=timezone.utc)
+
+
+# ── environment ──────────────────────────────────────────────────────────────
+
+
+def _env(tmp_path):
+    (tmp_path / "data").mkdir(exist_ok=True)
+    lib = tmp_path / "lib"
+    lib.mkdir(exist_ok=True)
+    archive = tmp_path / "archive"
+    archive.mkdir(exist_ok=True)
+    config = Config.from_env(
+        {
+            "PHOTO_PIPELINE_DATA_DIR": str(tmp_path / "data"),
+            "PHOTO_PIPELINE_LIBRARY_ROOTS": str(lib),
+            "PHOTO_PIPELINE_ARCHIVE_FREE_SPACE_RESERVE_BYTES": "0",
+        }
+    )
+    run_migrations(config.database_url)
+    return config, create_session_factory(create_db_engine(config.database_url)), lib, archive
+
+
+def structured(path, seed, size=(256, 192)):
+    """A deterministic, decodable photo — previews and pHashes must be real."""
+    path.parent.mkdir(parents=True, exist_ok=True)
+    rng = np.random.default_rng(seed)
+    w, h = size
+    base = np.zeros((h, w, 3), dtype=np.uint8)
+    for _ in range(6):
+        x0 = int(rng.integers(0, w - 60))
+        y0 = int(rng.integers(0, h - 60))
+        base[y0 : y0 + 60, x0 : x0 + 60] = rng.integers(0, 256, 3)
+    grad = np.linspace(0, 120, w, dtype=np.uint8)
+    base[:, :, 0] = np.clip(base[:, :, 0].astype(int) + grad[None, :], 0, 255)
+    Image.fromarray(base).save(path, quality=95)
+    return path
+
+
+def resized_copy(src, dst, scale=0.5):
+    with Image.open(src) as image:
+        image.resize(
+            (int(image.width * scale), int(image.height * scale)), Image.LANCZOS
+        ).save(dst, quality=95)
+    return dst
+
+
+def _uploaded(sf, lib, album="rome", seeds=(1, 2)):
+    """A scanned album carrying the verified upload evidence archiving requires."""
+    folder = lib / album
+    for index, seed in enumerate(seeds):
+        structured(folder / f"{index}.jpg", seed)
+    result = InventoryService(sf).scan(lib)
+    with sf() as session:
+        batch_id = str(uuid.uuid4())
+        session.add(
+            UploadBatch(
+                id=batch_id,
+                album=album,
+                folder=str(folder),
+                album_name=album,
+                state="succeeded",
+                preflight_token="v1:test",
+                outcome_state="verified",
+                created_at=NOW,
+            )
+        )
+        for path, asset_id in result.asset_ids.items():
+            session.add(
+                UploadItem(
+                    batch_id=batch_id,
+                    asset_id=asset_id,
+                    path=path,
+                    sha256=sha256_file(path),
+                    sha1="0" * 40,
+                    state="sent",
+                    outcome="uploaded",
+                )
+            )
+        session.commit()
+    return folder, result.asset_ids
+
+
+def _archive(sf, config, archive, albums=None):
+    service = ArchiveService(sf, config=config)
+    location = service.register("external", str(archive))
+    token = service.preflight(location["id"], albums)["token"]
+    transfers = ArchiveTransferService(sf, config=config)
+    plan = transfers.create(location["id"], albums, token=token)
+    transfers.apply(plan["id"])
+    return location
+
+
+def _unmount(archive):
+    """Take the medium away the way a real one goes: its marker stops answering."""
+    (archive / MARKER_NAME).rename(archive / f"{MARKER_NAME}.away")
+
+
+def _remount(archive):
+    (archive / f"{MARKER_NAME}.away").rename(archive / MARKER_NAME)
+
+
+def _assets(sf):
+    with sf() as session:
+        return {asset.id: asset for asset in session.scalars(select(Asset))}
+
+
+# ── availability ─────────────────────────────────────────────────────────────
+
+
+def test_archived_assets_report_online_offline_and_missing(tmp_path):
+    config, sf, lib, archive = _env(tmp_path)
+    _, ids = _uploaded(sf, lib)
+    _archive(sf, config, archive)
+    inventory = InventoryService(sf)
+
+    states = {a.availability_state for a in _assets(sf).values()}
+    assert states == {ARCHIVED_ONLINE}
+
+    _unmount(archive)
+    inventory.scan(lib)  # the album folder is gone from the active roots
+    assert {a.availability_state for a in _assets(sf).values()} == {ARCHIVED_OFFLINE}
+    assert all(a.missing_at is None for a in _assets(sf).values())  # not "missing"
+
+    _remount(archive)
+    inventory.scan(lib)
+    assert {a.availability_state for a in _assets(sf).values()} == {ARCHIVED_ONLINE}
+
+    # Mounted medium, absent file: that is not an offline archive, it needs a human.
+    victim = sorted(ids.values())[0]
+    with sf() as session:
+        asset = session.get(Asset, victim)
+        (archive / asset.archive_path).unlink()
+    inventory.scan(lib)
+    assert _assets(sf)[victim].availability_state == MISSING_UNEXPECTED
+
+
+def test_scan_never_prunes_or_flags_offline_assets(tmp_path):
+    config, sf, lib, archive = _env(tmp_path)
+    _, ids = _uploaded(sf, lib)
+    _archive(sf, config, archive)
+    _unmount(archive)
+
+    before = _assets(sf)
+    result = InventoryService(sf).scan(lib)
+
+    assert result.counts.get("missing") is None
+    after = _assets(sf)
+    assert set(after) == set(before) == set(ids.values())
+    for asset in after.values():
+        assert asset.availability_state == ARCHIVED_OFFLINE
+        assert asset.current_sha256 and asset.pixel_sha256  # hashes retained
+        assert asset.archive_location_id and asset.archive_path
+
+
+def test_active_missing_file_is_missing_unexpected_not_offline(tmp_path):
+    config, sf, lib, _archive_root = _env(tmp_path)
+    structured(lib / "loose" / "a.jpg", 7)
+    ids = InventoryService(sf).scan(lib).asset_ids
+    asset_id = next(iter(ids.values()))
+
+    (lib / "loose" / "a.jpg").unlink()
+    InventoryService(sf).scan(lib)
+
+    asset = _assets(sf)[asset_id]
+    assert asset.availability_state == MISSING_UNEXPECTED
+    assert asset.missing_at is not None
+
+
+# ── deduplication against archived originals ─────────────────────────────────
+
+
+def test_exact_copy_of_offline_asset_links_to_archived_canonical(tmp_path):
+    config, sf, lib, archive = _env(tmp_path)
+    folder, ids = _uploaded(sf, lib, seeds=(1,))
+    archived_id = next(iter(ids.values()))
+    original = next(iter(ids))
+    kept = tmp_path / "kept.jpg"
+    shutil.copy2(original, kept)
+    _archive(sf, config, archive)
+    _unmount(archive)
+
+    # The same photo turns up again in the active library while the disk is away.
+    (lib / "inbox").mkdir(parents=True, exist_ok=True)
+    shutil.copy2(kept, lib / "inbox" / "again.jpg")
+    scan = InventoryService(sf).scan(lib)
+    new_id = scan.asset_ids[str(lib / "inbox" / "again.jpg")]
+    assert scan.occurrences[str(lib / "inbox" / "again.jpg")] == "copied"
+
+    clusters = DuplicateService(sf).detect().clusters
+    exact = [c for c in clusters if c["method"] == Method.EXACT.value]
+    assert len(exact) == 1
+    cluster = exact[0]
+    # Byte-identical: linked directly, and the archived original stays canonical.
+    assert cluster["state"] == ClusterState.DECIDED.value
+    assert cluster["canonical_asset_id"] == archived_id
+    assert _assets(sf)[new_id].canonical_asset_id == archived_id
+
+
+def test_fuzzy_copy_of_offline_asset_requires_review_and_names_the_medium(tmp_path):
+    config, sf, lib, archive = _env(tmp_path)
+    folder, ids = _uploaded(sf, lib, seeds=(1,))
+    archived_id = next(iter(ids.values()))
+    original = next(iter(ids))
+    variant_source = resized_copy(original, tmp_path / "small.jpg")
+    _archive(sf, config, archive)
+    _unmount(archive)
+
+    (lib / "inbox").mkdir(parents=True, exist_ok=True)
+    shutil.copy2(variant_source, lib / "inbox" / "small.jpg")
+    InventoryService(sf).scan(lib)
+
+    duplicates = DuplicateService(sf)
+    clusters = duplicates.detect().clusters
+    perceptual = [c for c in clusters if c["method"] == Method.PERCEPTUAL.value]
+    assert len(perceptual) == 1
+
+    detail = duplicates.get_cluster(perceptual[0]["id"])
+    assert detail["state"] == ClusterState.OPEN.value  # never auto-decided
+    assert detail["requires_confirmation"] is True
+    assert detail["mount_required"] == ["external"]  # full-resolution needs the disk
+    archived = next(m for m in detail["members"] if m["asset_id"] == archived_id)
+    assert archived["availability_state"] == ARCHIVED_OFFLINE
+    assert archived["current_path"] is None
+    assert archived["archive_path"] and archived["evidence"]["phash"]
+    # The retained preview is what makes the offline member reviewable at all.
+    assert archived["preview"] == {"state": "ready", "protected": True, "size": PROTECTED_SIZE}
+
+
+# ── protected review evidence ────────────────────────────────────────────────
+
+
+def test_protected_preview_survives_quota_and_serves_while_offline(tmp_path):
+    config, sf, lib, archive = _env(tmp_path)
+    _, ids = _uploaded(sf, lib, seeds=(1,))
+    asset_id = next(iter(ids.values()))
+    _archive(sf, config, archive)
+    _unmount(archive)
+
+    thumbnails = ThumbnailService(sf, config)
+    served = thumbnails.generate(asset_id, PROTECTED_SIZE)
+    assert served.exists()  # rendered before the original left, not from the medium
+
+    # An aggressive quota may empty the cache, but not this evidence.
+    tight = ThumbnailService(sf, config.model_copy(update={"thumbnail_cache_quota_bytes": 1}))
+    tight._enforce_quota()
+    assert served.exists()
+    with sf() as session:
+        row = session.scalar(select(Thumbnail).where(Thumbnail.asset_id == asset_id))
+        assert row.protected is True
+    assert thumbnails.evidence(asset_id)["state"] == "ready"
+
+
+def test_offline_asset_without_preview_reports_unavailable(tmp_path):
+    """No preview and no medium is an honest 409, never a wrong picture."""
+    config, sf, lib, archive = _env(tmp_path)
+    _, ids = _uploaded(sf, lib, seeds=(1,))
+    asset_id = next(iter(ids.values()))
+    _archive(sf, config, archive)
+    with sf() as session:
+        for row in session.scalars(select(Thumbnail).where(Thumbnail.asset_id == asset_id)):
+            session.delete(row)
+        session.commit()
+    _unmount(archive)
+
+    with pytest.raises(ThumbnailUnavailable):
+        ThumbnailService(sf, config).generate(asset_id, 256)
+
+    _remount(archive)  # mounted again: the archived original is readable
+    assert ThumbnailService(sf, config).generate(asset_id, 256).exists()
+
+
+def test_offline_asset_is_browsable_through_the_api(tmp_path):
+    """The browser sees an archived asset, its medium, and its preview — offline."""
+    config, sf, lib, archive = _env(tmp_path)
+    _, ids = _uploaded(sf, lib, seeds=(1,))
+    asset_id = next(iter(ids.values()))
+    _archive(sf, config, archive)
+    _unmount(archive)
+
+    with TestClient(create_app(config)) as client:
+        client.post("/api/v1/inventory/scan")
+        listed = client.get("/api/v1/inventory/assets", params={"availability": ARCHIVED_OFFLINE})
+        assert listed.status_code == 200
+        item = next(row for row in listed.json()["items"] if row["id"] == asset_id)
+        assert item["current_path"] is None
+        assert item["archive_path"] == "rome/0.jpg"
+        assert item["missing"] is False
+
+        # Searching by the archived path still finds it.
+        found = client.get("/api/v1/inventory/assets", params={"q": "rome"}).json()
+        assert [row["id"] for row in found["items"]] == [asset_id]
+
+        preview = client.get(f"/api/v1/assets/{asset_id}/thumbnail", params={"size": 1280})
+        assert preview.status_code == 200
+        assert preview.headers["content-type"] == "image/webp"
+
+
+def test_offline_state_is_stable_across_restart(tmp_path):
+    config, sf, lib, archive = _env(tmp_path)
+    _, ids = _uploaded(sf, lib, seeds=(1, 2))
+    _archive(sf, config, archive)
+    _unmount(archive)
+    InventoryService(sf).scan(lib)
+    before = {
+        asset_id: (
+            asset.availability_state,
+            asset.archive_path,
+            asset.current_sha256,
+            asset.phash,
+        )
+        for asset_id, asset in _assets(sf).items()
+    }
+
+    # Restart: a fresh engine and session factory against the same database.
+    restarted = create_session_factory(create_db_engine(config.database_url))
+    after = {
+        asset_id: (
+            asset.availability_state,
+            asset.archive_path,
+            asset.current_sha256,
+            asset.phash,
+        )
+        for asset_id, asset in _assets(restarted).items()
+    }
+    assert after == before
+    assert set(after) == set(ids.values())
+
+    # And the medium coming back is picked up by the restarted process.
+    _remount(archive)
+    assert ArchiveService(restarted, config=config).locations()[0]["state"] == "online"
+    assert {a.availability_state for a in _assets(restarted).values()} == {ARCHIVED_ONLINE}
+    assert ACTIVE not in {a.availability_state for a in _assets(restarted).values()}

tests/integration/test_restore.py

@@ -0,0 +1,417 @@
+"""Planning and executing safe restores (US06-04).
+
+Restoring is the one archive operation that can *add* a file to the library, so
+every case here asks two questions: did the right bytes come back under the right
+identity, and did anything already in the library get touched? The media are real
+directories, the hashes are real, and the failure paths assert that the archived
+copy is still exactly where it was — a restore that fails must cost nothing.
+"""
+
+import shutil
+import uuid
+from datetime import datetime, timezone
+
+import numpy as np
+import pytest
+from fastapi.testclient import TestClient
+from PIL import Image
+from sqlalchemy import select
+
+from photo_pipeline.api.app import create_app
+from photo_pipeline.config import Config
+from photo_pipeline.db import create_db_engine, create_session_factory, run_migrations
+from photo_pipeline.models import Asset, AssetPath, SafetyReview, UploadBatch, UploadItem
+from photo_pipeline.services import availability
+from photo_pipeline.services.archive_journal import ArchiveState
+from photo_pipeline.services.archive_transfer import ArchiveTransferService
+from photo_pipeline.services.archives import MARKER_NAME, ArchiveError, ArchiveService
+from photo_pipeline.services.hashing import sha256_file
+from photo_pipeline.services.inventory import InventoryService
+from photo_pipeline.services.restores import RestoreService
+
+pytestmark = pytest.mark.phase_f  # part of the Phase F acceptance gate (US06-06)
+
+NOW = datetime(2026, 1, 1, tzinfo=timezone.utc)
+
+
+# ── environment ──────────────────────────────────────────────────────────────
+
+
+def _env(tmp_path):
+    (tmp_path / "data").mkdir(exist_ok=True)
+    lib = tmp_path / "lib"
+    lib.mkdir(exist_ok=True)
+    archive = tmp_path / "archive"
+    archive.mkdir(exist_ok=True)
+    config = Config.from_env(
+        {
+            "PHOTO_PIPELINE_DATA_DIR": str(tmp_path / "data"),
+            "PHOTO_PIPELINE_LIBRARY_ROOTS": str(lib),
+            "PHOTO_PIPELINE_ARCHIVE_FREE_SPACE_RESERVE_BYTES": "0",
+        }
+    )
+    run_migrations(config.database_url)
+    return config, create_session_factory(create_db_engine(config.database_url)), lib, archive
+
+
+def structured(path, seed, size=(192, 144)):
+    path.parent.mkdir(parents=True, exist_ok=True)
+    rng = np.random.default_rng(seed)
+    w, h = size
+    base = np.zeros((h, w, 3), dtype=np.uint8)
+    for _ in range(5):
+        x0 = int(rng.integers(0, w - 40))
+        y0 = int(rng.integers(0, h - 40))
+        base[y0 : y0 + 40, x0 : x0 + 40] = rng.integers(0, 256, 3)
+    Image.fromarray(base).save(path, quality=95)
+    return path
+
+
+def _archived(sf, config, lib, archive, album="rome", seeds=(1, 2)):
+    """A real album taken all the way through archiving, ready to be restored."""
+    folder = lib / album
+    for index, seed in enumerate(seeds):
+        structured(folder / f"{index}.jpg", seed)
+    scan = InventoryService(sf).scan(lib)
+    with sf() as session:
+        batch_id = str(uuid.uuid4())
+        session.add(
+            UploadBatch(
+                id=batch_id,
+                album=album,
+                folder=str(folder),
+                album_name=album,
+                state="succeeded",
+                preflight_token="v1:test",
+                outcome_state="verified",
+                created_at=NOW,
+            )
+        )
+        for path, asset_id in scan.asset_ids.items():
+            session.add(
+                UploadItem(
+                    batch_id=batch_id,
+                    asset_id=asset_id,
+                    path=path,
+                    sha256=sha256_file(path),
+                    sha1="0" * 40,
+                    state="sent",
+                    outcome="uploaded",
+                )
+            )
+            # A decision that must survive the whole round trip.
+            session.add(
+                SafetyReview(
+                    id=str(uuid.uuid4()),
+                    asset_id=asset_id,
+                    decision="sfw",
+                    score=0.01,
+                    reviewer="test",
+                )
+            )
+        session.commit()
+
+    service = ArchiveService(sf, config=config)
+    location = service.register("external", str(archive))
+    token = service.preflight(location["id"])["token"]
+    transfers = ArchiveTransferService(sf, config=config)
+    plan = transfers.create(location["id"], None, token=token)
+    transfers.apply(plan["id"])
+    return location, scan.asset_ids
+
+
+def _restore(sf, config, location_id, asset_ids=None):
+    service = RestoreService(sf, config=config)
+    token = service.preflight(location_id, asset_ids)["token"]
+    plan = service.create(location_id, asset_ids, token=token)
+    return service, plan, service.apply(plan["id"])
+
+
+def _unmount(archive):
+    (archive / MARKER_NAME).rename(archive / f"{MARKER_NAME}.away")
+
+
+def _assets(sf):
+    with sf() as session:
+        return {asset.id: asset for asset in session.scalars(select(Asset))}
+
+
+def _codes(report):
+    return {issue["code"] for issue in report["blockers"]} | {
+        issue["code"] for item in report["items"] for issue in item["blockers"]
+    }
+
+
+# ── preflight ────────────────────────────────────────────────────────────────
+
+
+def test_preflight_blocks_offline_medium(tmp_path):
+    config, sf, lib, archive = _env(tmp_path)
+    location, _ = _archived(sf, config, lib, archive)
+    _unmount(archive)
+
+    report = RestoreService(sf, config=config).preflight(location["id"])
+    assert report["state"] == "blocked"
+    assert "location_offline" in _codes(report)
+
+
+def test_preflight_blocks_wrong_volume(tmp_path):
+    config, sf, lib, archive = _env(tmp_path)
+    location, _ = _archived(sf, config, lib, archive)
+    (archive / MARKER_NAME).write_text('{"media_id": "someone-elses-disk"}', encoding="utf-8")
+
+    report = RestoreService(sf, config=config).preflight(location["id"])
+    assert report["state"] == "blocked"
+    assert "wrong_volume" in _codes(report)
+
+
+def test_preflight_blocks_changed_archive_bytes(tmp_path):
+    config, sf, lib, archive = _env(tmp_path)
+    location, ids = _archived(sf, config, lib, archive, seeds=(1,))
+    asset_id = next(iter(ids.values()))
+    with sf() as session:
+        archived_file = archive / session.get(Asset, asset_id).archive_path
+    archived_file.write_bytes(b"not the photo that was archived")
+
+    report = RestoreService(sf, config=config).preflight(location["id"])
+    assert report["state"] == "blocked"
+    assert "bytes_changed" in _codes(report)
+    with pytest.raises(ArchiveError) as error:
+        _restore(sf, config, location["id"])
+    assert error.value.code == "blocked"
+
+
+def test_preflight_blocks_insufficient_capacity(tmp_path):
+    config, sf, lib, archive = _env(tmp_path)
+    location, _ = _archived(sf, config, lib, archive)
+    greedy = config.model_copy(
+        update={"archive_free_space_reserve_bytes": 1 << 62}  # more than any disk has
+    )
+
+    report = RestoreService(sf, config=greedy).preflight(location["id"])
+    assert report["state"] == "blocked"
+    assert "insufficient_capacity" in _codes(report)
+
+
+def test_token_changes_with_the_scope(tmp_path):
+    config, sf, lib, archive = _env(tmp_path)
+    location, ids = _archived(sf, config, lib, archive)
+    service = RestoreService(sf, config=config)
+
+    whole = service.preflight(location["id"])["token"]
+    partial = service.preflight(location["id"], [sorted(ids.values())[0]])["token"]
+    assert whole != partial
+    assert service.verify_token(whole, location["id"])
+    assert not service.verify_token(partial, location["id"])
+
+
+# ── restore ──────────────────────────────────────────────────────────────────
+
+
+def test_restore_returns_bytes_identity_and_decisions(tmp_path):
+    config, sf, lib, archive = _env(tmp_path)
+    location, ids = _archived(sf, config, lib, archive)
+    archived_hashes = {
+        asset_id: asset.current_sha256 for asset_id, asset in _assets(sf).items()
+    }
+
+    service, plan, result = _restore(sf, config, location["id"])
+    assert (result["restored"], result["failed"], result["state"]) == (2, 0, "complete")
+
+    for asset_id, asset in _assets(sf).items():
+        assert asset.availability_state == availability.ACTIVE
+        assert asset.current_path == str(lib / asset.archive_path)
+        assert sha256_file(asset.current_path) == archived_hashes[asset_id]
+        # The archived copy is a copy: restoring never empties the medium.
+        assert (archive / asset.archive_path).exists()
+        assert asset.archive_location_id == location["id"]
+
+    with sf() as session:
+        # Identity and decisions survived: same ids, same reviews, new occurrence.
+        assert set(ids.values()) == {a.id for a in session.scalars(select(Asset))}
+        assert {r.decision for r in session.scalars(select(SafetyReview))} == {"sfw"}
+        occurrences = [
+            row.reason
+            for row in session.scalars(
+                select(AssetPath).where(AssetPath.asset_id == sorted(ids.values())[0])
+            )
+        ]
+        assert "restore" in occurrences
+
+
+def test_restore_never_overwrites_a_collision(tmp_path):
+    config, sf, lib, archive = _env(tmp_path)
+    location, ids = _archived(sf, config, lib, archive, seeds=(1,))
+    asset_id = next(iter(ids.values()))
+    with sf() as session:
+        archive_path = session.get(Asset, asset_id).archive_path
+    occupied = lib / archive_path
+    occupied.parent.mkdir(parents=True, exist_ok=True)
+    occupied.write_bytes(b"a different photo already lives here")
+    before = occupied.read_bytes()
+
+    service, plan, result = _restore(sf, config, location["id"])
+    assert result["failed"] == 0
+
+    assert occupied.read_bytes() == before  # untouched
+    restored = _assets(sf)[asset_id].current_path
+    assert restored != str(occupied)
+    assert "(restored)" in restored
+    assert sha256_file(restored) == sha256_file(archive / archive_path)
+
+
+def test_apply_refuses_a_destination_taken_after_planning(tmp_path):
+    config, sf, lib, archive = _env(tmp_path)
+    location, ids = _archived(sf, config, lib, archive, seeds=(1,))
+    service = RestoreService(sf, config=config)
+    token = service.preflight(location["id"])["token"]
+    plan = service.create(location["id"], None, token=token)
+
+    # Someone drops a file exactly where the plan intends to publish.
+    destination = plan["operations"][0]["destination_path"]
+    from pathlib import Path
+
+    Path(destination).parent.mkdir(parents=True, exist_ok=True)
+    Path(destination).write_bytes(b"squatter")
+
+    result = service.apply(plan["id"])
+    assert result["failed"] == 1
+    operation = service.journal.operations(plan["id"])[0]
+    assert operation["journal_state"] == ArchiveState.FAILED
+    assert operation["error_code"] == "destination_exists"
+    assert Path(destination).read_bytes() == b"squatter"
+    assert _assets(sf)[next(iter(ids.values()))].availability_state == availability.ARCHIVED_ONLINE
+
+
+def test_changed_archive_bytes_mark_the_asset_divergent(tmp_path):
+    config, sf, lib, archive = _env(tmp_path)
+    location, ids = _archived(sf, config, lib, archive, seeds=(1,))
+    asset_id = next(iter(ids.values()))
+    service = RestoreService(sf, config=config)
+    token = service.preflight(location["id"])["token"]
+    plan = service.create(location["id"], None, token=token)
+
+    # The medium's copy is edited after the plan was approved.
+    with sf() as session:
+        archived_file = archive / session.get(Asset, asset_id).archive_path
+    archived_file.write_bytes(b"edited on the shelf")
+
+    result = service.apply(plan["id"])
+    assert result["failed"] == 1
+    operation = service.journal.operations(plan["id"])[0]
+    assert operation["error_code"] == "bytes_changed"
+    asset = _assets(sf)[asset_id]
+    assert asset.archive_divergent_at is not None  # durable divergence
+    assert asset.availability_state == availability.ARCHIVED_ONLINE
+    assert asset.current_path is None  # nothing was published
+
+
+# ── interruption and idempotency ─────────────────────────────────────────────
+
+
+def test_interrupted_before_publishing_is_resumable(tmp_path):
+    config, sf, lib, archive = _env(tmp_path)
+    location, ids = _archived(sf, config, lib, archive, seeds=(1,))
+    service = RestoreService(sf, config=config)
+    token = service.preflight(location["id"])["token"]
+    plan = service.create(location["id"], None, token=token)
+    operation = service.journal.operations(plan["id"])[0]
+
+    # Model a kill right after the intent was written: nothing published yet.
+    service.journal.begin(operation["id"], worker_id="killed", fencing_token=1)
+
+    status = service.recovery_status()
+    assert status["operations"][0]["classification"] == "resumable"
+    assert service.recover() == {"resumed": 1, "completed": 0, "manual": 0}
+    assert service.journal.operations(plan["id"])[0]["journal_state"] == ArchiveState.PLANNED
+
+    result = service.apply(plan["id"])
+    assert result["failed"] == 0
+    assert _assets(sf)[next(iter(ids.values()))].availability_state == availability.ACTIVE
+
+
+def test_interrupted_after_publishing_is_finished_by_recovery(tmp_path):
+    config, sf, lib, archive = _env(tmp_path)
+    location, ids = _archived(sf, config, lib, archive, seeds=(1,))
+    asset_id = next(iter(ids.values()))
+    service = RestoreService(sf, config=config)
+    token = service.preflight(location["id"])["token"]
+    plan = service.create(location["id"], None, token=token)
+    operation = service.journal.operations(plan["id"])[0]
+
+    # Model a kill between the published copy and the database update.
+    from pathlib import Path
+
+    destination = Path(operation["destination_path"])
+    destination.parent.mkdir(parents=True, exist_ok=True)
+    shutil.copy2(operation["source_path"], destination)
+    service.journal.begin(operation["id"], worker_id="killed", fencing_token=1)
+    service.journal.transition(operation["id"], ArchiveState.VERIFIED, fencing_token=1)
+
+    assert service.recovery_status()["operations"][0]["classification"] == "forward"
+    assert service.recover()["completed"] == 1
+    asset = _assets(sf)[asset_id]
+    assert asset.availability_state == availability.ACTIVE
+    assert asset.current_path == str(destination)
+
+    # Repeated recovery and a repeated apply converge on the same state.
+    assert service.recover() == {"resumed": 0, "completed": 0, "manual": 0}
+    again = service.apply(plan["id"])
+    assert (again["skipped"], again["failed"]) == (1, 0)
+    assert _assets(sf)[asset_id].current_path == str(destination)
+
+
+def test_restored_state_survives_restart_and_rescan(tmp_path):
+    config, sf, lib, archive = _env(tmp_path)
+    location, ids = _archived(sf, config, lib, archive)
+    _restore(sf, config, location["id"])
+
+    restarted = create_session_factory(create_db_engine(config.database_url))
+    InventoryService(restarted).scan(lib)
+
+    assets = _assets(restarted)
+    assert set(assets) == set(ids.values())  # no new identities from the rescan
+    for asset in assets.values():
+        assert asset.availability_state == availability.ACTIVE
+        assert asset.missing_at is None
+
+    # Nothing is archived at that location any more, so there is nothing to restore.
+    again = RestoreService(restarted, config=config).preflight(location["id"])
+    assert _codes(again) == {"empty_scope"}
+
+
+def test_restore_api_round_trip(tmp_path):
+    config, sf, lib, archive = _env(tmp_path)
+    location, ids = _archived(sf, config, lib, archive, seeds=(1,))
+
+    with TestClient(create_app(config)) as client:
+        report = client.post(
+            "/api/v1/restore-preflight", json={"location_id": location["id"]}
+        ).json()
+        assert report["state"] == "ready"
+
+        stale = client.post(
+            "/api/v1/restore-plans",
+            json={"location_id": location["id"], "token": "r1:not-the-token"},
+        )
+        assert stale.status_code == 409
+
+        created = client.post(
+            "/api/v1/restore-plans",
+            json={"location_id": location["id"], "token": report["token"]},
+        )
+        assert created.status_code == 201
+        plan_id = created.json()["id"]
+        assert created.json()["direction"] == "restore"
+
+        # The plan is visible and applying it queues work on the archiver lane.
+        assert client.get(f"/api/v1/restore-plans/{plan_id}").status_code == 200
+        queued = client.post(f"/api/v1/restore-plans/{plan_id}/apply")
+        assert queued.status_code == 200
+        assert queued.json()["job"]["job_type"] == "restore_plan"
+        assert queued.json()["job"]["lock_key"] == "archive"
+        assert client.get("/api/v1/restore-recovery").json()["manual"] == []
+
+    assert _assets(sf)[next(iter(ids.values()))].availability_state == (
+        availability.ARCHIVED_ONLINE  # the worker, not the request, does the work
+    )

tests/story_traceability.json

@@ -123,6 +123,20 @@
     ],
     "US05-06": [
       "tests/e2e/test_phase_e_pipeline.py"
+    ],
+    "US06-01": [
+      "tests/integration/test_archive_preflight.py"
+    ],
+    "US06-02": [
+      "tests/unit/test_archive_journal_states.py",
+      "tests/integration/test_archive_transfer.py",
+      "tests/integration/test_archive_recovery.py"
+    ],
+    "US06-03": [
+      "tests/integration/test_offline_assets.py"
+    ],
+    "US06-04": [
+      "tests/integration/test_restore.py"
     ]
   }
 }

tests/unit/test_archive_journal_states.py

@@ -0,0 +1,105 @@
+"""Archive journal state machine and evidence table (US06-02).
+
+The evidence table decides whether an original may be deleted, so every
+combination of journal state and disk reality is asserted here as a pure function —
+no database, no files. A wrong cell in this table is data loss.
+"""
+
+import pytest
+
+from photo_pipeline.services.archive_journal import (
+    ALLOWED_TRANSITIONS,
+    FORWARD,
+    MANUAL,
+    RESUMABLE,
+    TERMINAL_STATES,
+    UNSAFE_STATES,
+    ArchiveState,
+    _classify,
+    can_transition,
+)
+
+pytestmark = pytest.mark.phase_f
+
+
+# ── state machine ────────────────────────────────────────────────────────────
+
+
+def test_the_happy_path_is_the_only_way_forward():
+    assert can_transition(ArchiveState.PLANNED, ArchiveState.TRANSFERRING)
+    assert can_transition(ArchiveState.TRANSFERRING, ArchiveState.VERIFIED)
+    assert can_transition(ArchiveState.VERIFIED, ArchiveState.REMOVING)
+    assert can_transition(ArchiveState.REMOVING, ArchiveState.COMPLETE)
+    # No shortcut may skip verification before a source is removed.
+    assert not can_transition(ArchiveState.TRANSFERRING, ArchiveState.REMOVING)
+    assert not can_transition(ArchiveState.PLANNED, ArchiveState.VERIFIED)
+    assert not can_transition(ArchiveState.VERIFIED, ArchiveState.COMPLETE)
+
+
+def test_removal_never_goes_backwards():
+    """Once the source may be gone, retrying the transfer would archive nothing and
+    could overwrite the copy that is now the only one."""
+    assert ALLOWED_TRANSITIONS[ArchiveState.REMOVING] == {
+        ArchiveState.COMPLETE,
+        ArchiveState.FAILED,
+    }
+    assert not can_transition(ArchiveState.REMOVING, ArchiveState.TRANSFERRING)
+    assert not can_transition(ArchiveState.REMOVING, ArchiveState.PLANNED)
+
+
+def test_complete_is_terminal_and_failed_can_be_retried():
+    assert ALLOWED_TRANSITIONS[ArchiveState.COMPLETE] == set()
+    assert TERMINAL_STATES == {ArchiveState.COMPLETE}
+    assert can_transition(ArchiveState.FAILED, ArchiveState.TRANSFERRING)
+    assert can_transition(ArchiveState.FAILED, ArchiveState.PLANNED)
+
+
+def test_every_state_that_can_touch_the_disk_is_marked_unsafe():
+    assert UNSAFE_STATES == {
+        ArchiveState.TRANSFERRING,
+        ArchiveState.VERIFIED,
+        ArchiveState.REMOVING,
+    }
+    assert ArchiveState.PLANNED not in UNSAFE_STATES
+
+
+# ── evidence table ───────────────────────────────────────────────────────────
+
+
+@pytest.mark.parametrize(
+    "state,source,destination,matches,expected",
+    [
+        # Nothing published yet: the source is still the only copy.
+        (ArchiveState.TRANSFERRING, True, False, False, RESUMABLE),
+        (ArchiveState.FAILED, True, False, False, RESUMABLE),
+        # The archive copy is durable and correct: finish the remaining steps.
+        (ArchiveState.TRANSFERRING, True, True, True, FORWARD),
+        (ArchiveState.TRANSFERRING, False, True, True, FORWARD),
+        (ArchiveState.VERIFIED, True, True, True, FORWARD),
+        (ArchiveState.REMOVING, False, True, True, FORWARD),
+        (ArchiveState.FAILED, True, True, True, FORWARD),
+        # Wrong bytes at the destination: never overwrite, never remove.
+        (ArchiveState.TRANSFERRING, True, True, False, MANUAL),
+        (ArchiveState.VERIFIED, True, True, False, MANUAL),
+        (ArchiveState.REMOVING, False, True, False, MANUAL),
+        # The journal claims an archived copy that is not there.
+        (ArchiveState.VERIFIED, True, False, False, MANUAL),
+        (ArchiveState.REMOVING, False, False, False, MANUAL),
+        # Neither copy exists — never silently accepted as success.
+        (ArchiveState.TRANSFERRING, False, False, False, MANUAL),
+        (ArchiveState.FAILED, False, False, False, MANUAL),
+    ],
+)
+def test_classification_of_every_evidence_combination(
+    state, source, destination, matches, expected
+):
+    classification, reason = _classify(state, source, destination, matches)
+
+    assert classification == expected, reason
+
+
+def test_a_source_that_is_gone_without_an_archive_copy_is_never_called_recoverable():
+    """The one combination that must always stop: the original left active storage
+    and nothing verifiable took its place."""
+    for state in (ArchiveState.TRANSFERRING, ArchiveState.VERIFIED, ArchiveState.REMOVING):
+        assert _classify(state, False, False, False)[0] == MANUAL