migrations/versions/0007_rename_plans.py

@@ -0,0 +1,100 @@
+"""Rename plans and their per-operation journal (US04-01, US04-02).
+
+Revision ID: 0007_rename_plans
+Revises: 0006_album_proposals
+Create Date: 2026-08-15
+
+The journal columns (state machine, attempts, fencing token, verification evidence)
+are created here alongside the plan itself so the crash-safe journal of US04-02 does
+not have to re-migrate the very rows US04-01 creates. US04-01 fills the planning and
+validation columns; the journal columns stay at their initial values until apply.
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = "0007_rename_plans"
+down_revision = "0006_album_proposals"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "rename_plans",
+        sa.Column("id", sa.String(), primary_key=True),
+        # planned | validated | invalid | applying | applied | failed | rolled_back
+        sa.Column("state", sa.String(), nullable=False, server_default="planned"),
+        sa.Column("schema_version", sa.Integer(), nullable=False, server_default="1"),
+        # Content checksum over the plan's operations; a portable export carries it
+        # so a stale confirmation can be detected without trusting the browser.
+        sa.Column("checksum", sa.String(), nullable=True),
+        sa.Column("blockers", sa.String(), nullable=True),  # JSON array of issues
+        sa.Column("operation_count", sa.Integer(), nullable=False, server_default="0"),
+        # Optimistic concurrency for confirmation (apply must present this).
+        sa.Column("version", sa.Integer(), nullable=False, server_default="1"),
+        sa.Column("validated_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("applied_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column(
+            "created_at",
+            sa.DateTime(timezone=True),
+            nullable=False,
+            server_default=sa.text("CURRENT_TIMESTAMP"),
+        ),
+        sa.Column(
+            "updated_at",
+            sa.DateTime(timezone=True),
+            nullable=False,
+            server_default=sa.text("CURRENT_TIMESTAMP"),
+        ),
+    )
+    op.create_index("ix_rename_plans_state", "rename_plans", ["state"])
+
+    op.create_table(
+        "rename_operations",
+        sa.Column("id", sa.String(), primary_key=True),
+        sa.Column(
+            "plan_id",
+            sa.String(),
+            sa.ForeignKey("rename_plans.id", ondelete="CASCADE"),
+            nullable=False,
+        ),
+        # Deterministic order within the plan (parents before children).
+        sa.Column("sequence", sa.Integer(), nullable=False),
+        # move_folder | move_file — folder moves are the default (album renames).
+        sa.Column("operation", sa.String(), nullable=False),
+        sa.Column("album", sa.String(), nullable=True),
+        sa.Column("source_path", sa.String(), nullable=False),
+        sa.Column("destination_path", sa.String(), nullable=False),
+        # Preconditions captured at planning time and rechecked before mutation.
+        sa.Column("asset_ids", sa.String(), nullable=True),  # JSON array
+        sa.Column("expected_sha256", sa.String(), nullable=True),  # JSON {asset_id: sha}
+        sa.Column("asset_count", sa.Integer(), nullable=False, server_default="0"),
+        sa.Column("same_filesystem", sa.Boolean(), nullable=True),
+        sa.Column("case_only", sa.Boolean(), nullable=False, server_default="0"),
+        sa.Column("issues", sa.String(), nullable=True),  # JSON array of issues
+        # ── journal (US04-02): untouched until apply ─────────────────────────
+        # planned → validated → moving → moved → database_updated → verified
+        #        → complete, or failed / rollback_required
+        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("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.text("CURRENT_TIMESTAMP"),
+        ),
+        sa.UniqueConstraint("plan_id", "sequence", name="uq_rename_operations_plan_sequence"),
+    )
+    op.create_index("ix_rename_operations_plan_id", "rename_operations", ["plan_id"])
+    op.create_index("ix_rename_operations_journal_state", "rename_operations", ["journal_state"])
+
+
+def downgrade() -> None:
+    op.drop_table("rename_operations")
+    op.drop_table("rename_plans")

photo_pipeline/api/app.py

@@ -22,6 +22,7 @@ from photo_pipeline.api.routes import (
     inventory,
     jobs,
     library,
+    renames,
     safety,
     thumbnails,
     workflow,
@@ -65,6 +66,7 @@ def create_app(config: Config | None = None) -> FastAPI:
     app.include_router(analysis.router, prefix="/api/v1")
     app.include_router(library.router, prefix="/api/v1")
     app.include_router(albums.router, prefix="/api/v1")
+    app.include_router(renames.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/renames.py

@@ -0,0 +1,54 @@
+"""Rename plan API (US04-01): build, inspect, and export plans.
+
+Planning is read-only — it previews and validates but never touches the filesystem.
+Applying a plan is a later story, so there is deliberately no apply endpoint here.
+"""
+
+from __future__ import annotations
+
+from fastapi import APIRouter, Request
+from fastapi.responses import JSONResponse
+
+from photo_pipeline.services.renames import RenameError, RenameService
+
+router = APIRouter(tags=["renames"])
+
+
+def _service(request: Request) -> RenameService:
+    return RenameService(
+        request.app.state.session_factory,
+        library_roots=tuple(request.app.state.config.library_roots),
+    )
+
+
+def _error(status: int, code: str, message: str) -> JSONResponse:
+    return JSONResponse(status_code=status, content={"error": {"code": code, "message": message}})
+
+
+@router.post("/rename-plans")
+def build_plan(request: Request):
+    try:
+        return _service(request).build_plan()
+    except RenameError as error:
+        return _error(422, "nothing_to_plan", str(error))
+
+
+@router.get("/rename-plans")
+def list_plans(request: Request) -> dict:
+    return _service(request).list_plans()
+
+
+@router.get("/rename-plans/{plan_id}")
+def get_plan(plan_id: str, request: Request):
+    plan = _service(request).get(plan_id)
+    if plan is None:
+        return _error(404, "not_found", f"unknown rename plan {plan_id}")
+    return plan
+
+
+@router.get("/rename-plans/{plan_id}/export")
+def export_plan(plan_id: str, request: Request):
+    try:
+        return _service(request).export(plan_id)
+    except RenameError as error:
+        return _error(404, "not_found", str(error))

photo_pipeline/models/__init__.py

@@ -12,6 +12,7 @@ from photo_pipeline.models.duplicates import (
     DuplicateNegativeLink,
 )
 from photo_pipeline.models.jobs import Job, JobEvent, JobItem
+from photo_pipeline.models.renames import RenameOperation, RenamePlan
 from photo_pipeline.models.thumbnails import Thumbnail
 from photo_pipeline.models.workflow import AnalysisResult, SafetyReview
 
@@ -25,6 +26,8 @@ __all__ = [
     "Job",
     "JobItem",
     "JobEvent",
+    "RenamePlan",
+    "RenameOperation",
     "Thumbnail",
     "SafetyReview",
     "AnalysisResult",

photo_pipeline/models/renames.py

@@ -0,0 +1,84 @@
+"""Rename plan and per-operation journal persistence (US04-01, US04-02).
+
+A plan is the complete, reviewable preview of a set of folder renames: every source,
+destination, affected asset, expected hash, and validation issue. Nothing here
+touches the filesystem — applying a plan is US04-03.
+
+The journal columns on ``RenameOperation`` belong to US04-02's crash-safe state
+machine and are created with the table so the journal never has to re-migrate the
+rows the planner writes. They stay at their defaults until apply runs.
+"""
+
+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 RenamePlan(Base):
+    __tablename__ = "rename_plans"
+
+    id: Mapped[str] = mapped_column(String, primary_key=True)
+    # planned | validated | invalid | applying | applied | failed | rolled_back
+    state: Mapped[str] = mapped_column(String, nullable=False, default="planned")
+    schema_version: Mapped[int] = mapped_column(Integer, nullable=False, default=1)
+    checksum: Mapped[str | None] = mapped_column(String)
+    blockers: Mapped[str | None] = mapped_column(String)  # JSON array
+    operation_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
+    version: Mapped[int] = mapped_column(Integer, nullable=False, default=1)
+    validated_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+    applied_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 RenameOperation(Base):
+    __tablename__ = "rename_operations"
+    __table_args__ = (
+        UniqueConstraint("plan_id", "sequence", name="uq_rename_operations_plan_sequence"),
+    )
+
+    id: Mapped[str] = mapped_column(String, primary_key=True)
+    plan_id: Mapped[str] = mapped_column(
+        ForeignKey("rename_plans.id", ondelete="CASCADE"), nullable=False, index=True
+    )
+    sequence: Mapped[int] = mapped_column(Integer, nullable=False)
+    operation: Mapped[str] = mapped_column(String, nullable=False)
+    album: Mapped[str | None] = mapped_column(String)
+    source_path: Mapped[str] = mapped_column(String, nullable=False)
+    destination_path: Mapped[str] = mapped_column(String, nullable=False)
+
+    asset_ids: Mapped[str | None] = mapped_column(String)  # JSON array
+    expected_sha256: Mapped[str | None] = mapped_column(String)  # JSON {asset_id: sha}
+    asset_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
+    same_filesystem: Mapped[bool | None] = mapped_column(Boolean)
+    case_only: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
+    issues: Mapped[str | None] = mapped_column(String)  # JSON array
+
+    # ── journal (US04-02) ────────────────────────────────────────────────────
+    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))
+    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/services/renames.py

@@ -0,0 +1,426 @@
+"""RenameService — build, validate, and export guarded rename plans (US04-01).
+
+A plan turns approved album proposals into an itemized, reviewable preview: for every
+album, the source folder, the destination folder, the assets that move with it, their
+expected hashes, and every validation issue found. **Nothing here touches the
+filesystem** — it only reads to validate. Applying a plan is US04-03.
+
+Validation is the point of the story, so every refusal is a structured issue rather
+than an exception (concept §7 "guarded rename plan"):
+
+- ``source_missing``      — the registered folder is gone
+- ``source_changed``      — an asset's bytes changed since inventory
+- ``duplicate_target``    — two operations claim the same destination
+- ``destination_exists``  — an unexpected path already occupies the destination
+- ``case_only``           — informational: needs the staged procedure on
+                            case-insensitive filesystems (not a blocker)
+- ``unicode_collision``   — two destinations differ only by Unicode normalization
+- ``root_escape``         — a path resolves outside the configured library
+- ``excluded_path``       — a path lies under ``_IGNORE/``
+- ``symlink``             — source or destination is a symlink
+- ``cross_filesystem``    — destination is on another device (copy-verify-delete)
+
+A plan with any blocking issue is ``invalid`` and can never be applied. Issues that
+are purely informational (``case_only``, ``cross_filesystem``) are recorded on the
+operation so the apply stage picks the right procedure.
+
+The JSON export is portable evidence: schema version, checksum, and the full operation
+list, with no credentials and no absolute host paths beyond the library itself.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import os
+import unicodedata
+import uuid
+from datetime import datetime, timezone
+from pathlib import Path
+
+from sqlalchemy import select
+from sqlalchemy.orm import sessionmaker
+
+from photo_pipeline import path_policy
+from photo_pipeline.models import AlbumProposal, Asset, RenameOperation, RenamePlan
+from photo_pipeline.services.albums import album_label
+
+EXPORT_SCHEMA_VERSION = 1
+MOVE_FOLDER = "move_folder"
+
+# Issues that make a plan unapplyable. Everything else is advisory.
+BLOCKING_CODES = frozenset(
+    {
+        "source_missing",
+        "source_changed",
+        "duplicate_target",
+        "destination_exists",
+        "unicode_collision",
+        "root_escape",
+        "excluded_path",
+        "symlink",
+        "no_library_root",
+    }
+)
+
+
+class RenameError(RuntimeError):
+    """Invalid request against a plan (unknown id, nothing to plan)."""
+
+
+def _now() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+def _issue(code: str, message: str) -> dict:
+    return {"code": code, "message": message}
+
+
+def _nfc(value: str) -> str:
+    return unicodedata.normalize("NFC", value)
+
+
+def plan_checksum(operations: list[dict]) -> str:
+    """Content hash over the operations that define a plan. Two plans with the same
+    moves have the same checksum, so a stale confirmation is detectable."""
+    payload = json.dumps(
+        [
+            {
+                "sequence": op["sequence"],
+                "source_path": op["source_path"],
+                "destination_path": op["destination_path"],
+                "asset_ids": sorted(op["asset_ids"]),
+            }
+            for op in sorted(operations, key=lambda op: op["sequence"])
+        ],
+        ensure_ascii=False,
+        sort_keys=True,
+    )
+    return hashlib.sha256(payload.encode("utf-8")).hexdigest()
+
+
+class RenameService:
+    def __init__(self, session_factory: sessionmaker, *, library_roots: tuple = ()) -> None:
+        self._session_factory = session_factory
+        self._roots = tuple(Path(root) for root in library_roots)
+
+    # ── planning ──────────────────────────────────────────────────────────────
+
+    def build_plan(self) -> dict:
+        """Build a plan from every approved album proposal. Read-only."""
+        with self._session_factory() as session:
+            approved = list(
+                session.scalars(
+                    select(AlbumProposal)
+                    .where(AlbumProposal.status == "approved")
+                    .order_by(AlbumProposal.album)
+                )
+            )
+            if not approved:
+                raise RenameError("no approved album proposals to plan")
+            assets = list(
+                session.scalars(
+                    select(Asset).where(
+                        Asset.canonical_asset_id.is_(None),
+                        Asset.availability_state == "active",
+                        Asset.current_path.is_not(None),
+                    )
+                )
+            )
+
+        by_album: dict[str, list[Asset]] = {}
+        for asset in assets:
+            by_album.setdefault(album_label(asset.current_path, self._roots), []).append(asset)
+
+        operations: list[dict] = []
+        for sequence, proposal in enumerate(approved):
+            members = sorted(by_album.get(proposal.album, []), key=lambda a: a.current_path)
+            operations.append(self._operation(sequence, proposal, members))
+
+        self._detect_cross_operation_issues(operations)
+        return self._persist(operations)
+
+    def _operation(self, sequence: int, proposal: AlbumProposal, members: list[Asset]) -> dict:
+        final_name = proposal.final_name or proposal.proposed_name or ""
+        issues: list[dict] = []
+
+        source = self._source_folder(members)
+        destination = source.parent / final_name if source else None
+
+        operation = {
+            "sequence": sequence,
+            "operation": MOVE_FOLDER,
+            "album": proposal.album,
+            "source_path": str(source) if source else "",
+            "destination_path": str(destination) if destination else "",
+            "asset_ids": [asset.id for asset in members],
+            "expected_sha256": {
+                asset.id: asset.current_sha256 for asset in members if asset.current_sha256
+            },
+            "asset_count": len(members),
+            "same_filesystem": None,
+            "case_only": False,
+            "issues": issues,
+        }
+
+        if source is None:
+            issues.append(
+                _issue("source_missing", f"no active assets remain in {proposal.album!r}")
+            )
+            return operation
+        if not final_name:
+            issues.append(_issue("empty_name", f"proposal for {proposal.album!r} has no name"))
+            return operation
+
+        self._validate_paths(operation, source, destination, issues)
+        self._validate_sources(operation, members, issues)
+        return operation
+
+    def _source_folder(self, members: list[Asset]) -> Path | None:
+        if not members:
+            return None
+        return Path(members[0].current_path).parent
+
+    def _validate_paths(
+        self, operation: dict, source: Path, destination: Path, issues: list[dict]
+    ) -> None:
+        if not self._roots:
+            issues.append(_issue("no_library_root", "no library root is configured"))
+            return
+
+        for label, path in (("source", source), ("destination", destination)):
+            if path_policy.is_excluded(path):
+                issues.append(_issue("excluded_path", f"{label} is under an excluded directory"))
+            root = self._root_for(path)
+            if root is None:
+                issues.append(
+                    _issue("root_escape", f"{label} {path} is outside every library root")
+                )
+
+        if source.is_symlink() or destination.is_symlink():
+            issues.append(_issue("symlink", "refusing to rename through a symlink"))
+
+        if not source.exists():
+            issues.append(_issue("source_missing", f"source folder {source} does not exist"))
+            return
+
+        # A symlinked source could still resolve outside the root even when its
+        # literal path looks fine; check the resolved location too.
+        root = self._root_for(source)
+        if root is not None:
+            try:
+                path_policy.resolve_within(root, source)
+            except path_policy.PathPolicyError as error:
+                issues.append(_issue("root_escape", str(error)))
+
+        operation["case_only"] = (
+            source != destination and source.as_posix().lower() == destination.as_posix().lower()
+        )
+        operation["same_filesystem"] = self._same_filesystem(source, destination)
+        if operation["same_filesystem"] is False:
+            issues.append(
+                _issue(
+                    "cross_filesystem", "destination is on another filesystem; copy-verify-delete"
+                )
+            )
+
+        # A destination that already exists is only acceptable for a case-only
+        # rename, where source and destination are the same directory entry.
+        if destination.exists() and not operation["case_only"]:
+            issues.append(_issue("destination_exists", f"destination {destination} already exists"))
+
+    def _validate_sources(self, operation: dict, members: list[Asset], issues: list[dict]) -> None:
+        """Every asset must still be present with the bytes inventory recorded."""
+        for asset in members:
+            path = Path(asset.current_path)
+            if not path.exists():
+                issues.append(_issue("source_missing", f"asset file {path} is missing"))
+                continue
+            expected = asset.byte_size
+            if expected is not None and path.stat().st_size != expected:
+                issues.append(
+                    _issue("source_changed", f"asset file {path} changed since inventory")
+                )
+
+    def _root_for(self, path: Path) -> Path | None:
+        for root in self._roots:
+            resolved_root = Path(root)
+            if path == resolved_root or resolved_root in path.parents:
+                return resolved_root
+        return None
+
+    def _same_filesystem(self, source: Path, destination: Path) -> bool | None:
+        anchor = destination.parent
+        if not source.exists() or not anchor.exists():
+            return None
+        try:
+            return os.stat(source).st_dev == os.stat(anchor).st_dev
+        except OSError:
+            return None
+
+    def _detect_cross_operation_issues(self, operations: list[dict]) -> None:
+        """Two operations must never target the same destination — including when the
+        names differ only by case or by Unicode normalization."""
+        exact: dict[str, int] = {}
+        folded: dict[str, int] = {}
+        for operation in operations:
+            destination = operation["destination_path"]
+            if not destination:
+                continue
+            if destination in exact:
+                operation["issues"].append(
+                    _issue("duplicate_target", f"destination {destination} is claimed twice")
+                )
+            else:
+                exact[destination] = operation["sequence"]
+
+            key = _nfc(destination).casefold()
+            if key in folded and folded[key] != operation["sequence"]:
+                operation["issues"].append(
+                    _issue(
+                        "unicode_collision",
+                        f"destination {destination} collides with another target after "
+                        "case/Unicode normalization",
+                    )
+                )
+            else:
+                folded.setdefault(key, operation["sequence"])
+
+    def _persist(self, operations: list[dict]) -> dict:
+        blockers = sorted(
+            {
+                issue["code"]
+                for operation in operations
+                for issue in operation["issues"]
+                if issue["code"] in BLOCKING_CODES
+            }
+        )
+        now = _now()
+        plan_id = str(uuid.uuid4())
+        with self._session_factory() as session:
+            session.add(
+                RenamePlan(
+                    id=plan_id,
+                    state="invalid" if blockers else "validated",
+                    schema_version=EXPORT_SCHEMA_VERSION,
+                    checksum=plan_checksum(operations),
+                    blockers=json.dumps(blockers),
+                    operation_count=len(operations),
+                    version=1,
+                    validated_at=now,
+                )
+            )
+            # Flush the parent before its operations so the foreign key resolves.
+            session.flush()
+            for operation in operations:
+                session.add(
+                    RenameOperation(
+                        id=str(uuid.uuid4()),
+                        plan_id=plan_id,
+                        sequence=operation["sequence"],
+                        operation=operation["operation"],
+                        album=operation["album"],
+                        source_path=operation["source_path"],
+                        destination_path=operation["destination_path"],
+                        asset_ids=json.dumps(operation["asset_ids"]),
+                        expected_sha256=json.dumps(operation["expected_sha256"]),
+                        asset_count=operation["asset_count"],
+                        same_filesystem=operation["same_filesystem"],
+                        case_only=operation["case_only"],
+                        issues=json.dumps(operation["issues"], ensure_ascii=False),
+                    )
+                )
+            session.commit()
+        return self.get(plan_id)
+
+    # ── reads ─────────────────────────────────────────────────────────────────
+
+    def get(self, plan_id: str) -> dict | None:
+        with self._session_factory() as session:
+            plan = session.get(RenamePlan, plan_id)
+            if plan is None:
+                return None
+            operations = list(
+                session.scalars(
+                    select(RenameOperation)
+                    .where(RenameOperation.plan_id == plan_id)
+                    .order_by(RenameOperation.sequence)
+                )
+            )
+            return _plan_dict(plan, operations)
+
+    def list_plans(self) -> dict:
+        with self._session_factory() as session:
+            plans = list(session.scalars(select(RenamePlan).order_by(RenamePlan.created_at.desc())))
+            return {
+                "items": [
+                    {
+                        "id": plan.id,
+                        "state": plan.state,
+                        "operation_count": plan.operation_count,
+                        "blockers": json.loads(plan.blockers or "[]"),
+                        "checksum": plan.checksum,
+                        "version": plan.version,
+                    }
+                    for plan in plans
+                ],
+                "total": len(plans),
+            }
+
+    def export(self, plan_id: str) -> dict:
+        """Portable JSON evidence for a plan. Contains no credentials."""
+        plan = self.get(plan_id)
+        if plan is None:
+            raise RenameError(f"unknown rename plan {plan_id!r}")
+        return {
+            "schema_version": plan["schema_version"],
+            "plan_id": plan["id"],
+            "state": plan["state"],
+            "checksum": plan["checksum"],
+            "blockers": plan["blockers"],
+            "operations": [
+                {
+                    "sequence": operation["sequence"],
+                    "operation": operation["operation"],
+                    "album": operation["album"],
+                    "source_path": operation["source_path"],
+                    "destination_path": operation["destination_path"],
+                    "asset_ids": operation["asset_ids"],
+                    "asset_count": operation["asset_count"],
+                    "case_only": operation["case_only"],
+                    "same_filesystem": operation["same_filesystem"],
+                    "issues": operation["issues"],
+                }
+                for operation in plan["operations"]
+            ],
+        }
+
+
+def _plan_dict(plan: RenamePlan, operations: list[RenameOperation]) -> dict:
+    return {
+        "id": plan.id,
+        "state": plan.state,
+        "schema_version": plan.schema_version,
+        "checksum": plan.checksum,
+        "blockers": json.loads(plan.blockers or "[]"),
+        "operation_count": plan.operation_count,
+        "version": plan.version,
+        "applicable": plan.state == "validated",
+        "operations": [
+            {
+                "sequence": operation.sequence,
+                "operation": operation.operation,
+                "album": operation.album,
+                "source_path": operation.source_path,
+                "destination_path": operation.destination_path,
+                "asset_ids": json.loads(operation.asset_ids or "[]"),
+                "expected_sha256": json.loads(operation.expected_sha256 or "{}"),
+                "asset_count": operation.asset_count,
+                "same_filesystem": operation.same_filesystem,
+                "case_only": operation.case_only,
+                "issues": json.loads(operation.issues or "[]"),
+                "journal_state": operation.journal_state,
+            }
+            for operation in operations
+        ],
+    }

tests/integration/test_rename_plans.py

@@ -0,0 +1,412 @@
+"""Rename plan building, validation, and export (US04-01).
+
+Plans are built against a real temporary library so path validation (missing sources,
+collisions, case-only, Unicode, root escapes, symlinks) is exercised against real
+filesystem behaviour. Planning must never mutate anything: every test that builds a
+plan also asserts the library is byte-for-byte unchanged.
+"""
+
+import json
+import random
+import unicodedata
+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 AlbumProposal, Asset
+from photo_pipeline.services.renames import (
+    BLOCKING_CODES,
+    RenameError,
+    RenameService,
+    plan_checksum,
+)
+
+NOW = datetime(2026, 1, 1, tzinfo=timezone.utc)
+
+
+def _factory(tmp_path):
+    (tmp_path / "data").mkdir()
+    lib = tmp_path / "lib"
+    lib.mkdir()
+    config = Config.from_env(
+        {
+            "PHOTO_PIPELINE_DATA_DIR": str(tmp_path / "data"),
+            "PHOTO_PIPELINE_LIBRARY_ROOTS": str(lib),
+        }
+    )
+    run_migrations(config.database_url)
+    return config, create_session_factory(create_db_engine(config.database_url)), lib
+
+
+def _album(sf, lib, album, names=("a.jpg", "b.jpg"), *, approved_name="2019 Rome"):
+    """Create a real album folder with files, registered assets, and an approved
+    proposal naming it ``approved_name``."""
+    folder = lib / album
+    folder.mkdir(parents=True, exist_ok=True)
+    with sf() as session:
+        for name in names:
+            path = folder / name
+            path.write_bytes(b"x" * 32)
+            session.add(
+                Asset(
+                    id=str(uuid.uuid4()),
+                    original_path=str(path),
+                    current_path=str(path),
+                    discovered_at=NOW,
+                    hash_version=1,
+                    byte_size=32,
+                    current_sha256="deadbeef",
+                )
+            )
+        session.add(
+            AlbumProposal(
+                id=str(uuid.uuid4()),
+                album=album,
+                proposed_name=approved_name,
+                final_name=approved_name,
+                status="approved",
+                version=2,
+            )
+        )
+        session.commit()
+    return folder
+
+
+def _snapshot(lib):
+    """Every path under the library plus its bytes — proof planning mutates nothing."""
+    return {
+        str(p.relative_to(lib)): (p.read_bytes() if p.is_file() else None)
+        for p in sorted(lib.rglob("*"))
+    }
+
+
+def _service(sf, lib):
+    return RenameService(sf, library_roots=(lib,))
+
+
+def _codes(plan):
+    return {issue["code"] for op in plan["operations"] for issue in op["issues"]}
+
+
+# ── happy path ───────────────────────────────────────────────────────────────
+
+
+def test_plan_lists_sources_destinations_assets_and_hashes(tmp_path):
+    _, sf, lib = _factory(tmp_path)
+    _album(sf, lib, "rome", approved_name="2019 Rome")
+    before = _snapshot(lib)
+
+    plan = _service(sf, lib).build_plan()
+
+    assert plan["state"] == "validated" and plan["applicable"] is True
+    assert plan["operation_count"] == 1 and plan["blockers"] == []
+    operation = plan["operations"][0]
+    assert operation["source_path"] == str(lib / "rome")
+    assert operation["destination_path"] == str(lib / "2019 Rome")
+    assert operation["asset_count"] == 2 and len(operation["asset_ids"]) == 2
+    assert set(operation["expected_sha256"].values()) == {"deadbeef"}
+    assert operation["journal_state"] == "planned"  # journal untouched until apply
+    assert _snapshot(lib) == before, "planning must not touch the filesystem"
+
+
+def test_plans_without_approved_proposals_are_refused(tmp_path):
+    _, sf, lib = _factory(tmp_path)
+    with pytest.raises(RenameError):
+        _service(sf, lib).build_plan()
+
+
+# ── validation goldens ───────────────────────────────────────────────────────
+
+
+def test_missing_source_folder_blocks_the_plan(tmp_path):
+    _, sf, lib = _factory(tmp_path)
+    folder = _album(sf, lib, "rome")
+    for child in folder.iterdir():
+        child.unlink()
+    folder.rmdir()
+
+    plan = _service(sf, lib).build_plan()
+    assert plan["state"] == "invalid" and plan["applicable"] is False
+    assert "source_missing" in _codes(plan)
+
+
+def test_changed_source_bytes_block_the_plan(tmp_path):
+    _, sf, lib = _factory(tmp_path)
+    folder = _album(sf, lib, "rome")
+    (folder / "a.jpg").write_bytes(b"different length entirely")
+
+    plan = _service(sf, lib).build_plan()
+    assert "source_changed" in _codes(plan) and plan["state"] == "invalid"
+
+
+def test_existing_destination_is_never_overwritten(tmp_path):
+    _, sf, lib = _factory(tmp_path)
+    _album(sf, lib, "rome", approved_name="2019 Rome")
+    (lib / "2019 Rome").mkdir()  # something already lives there
+
+    plan = _service(sf, lib).build_plan()
+    assert "destination_exists" in _codes(plan) and plan["state"] == "invalid"
+
+
+def test_two_albums_targeting_one_destination_collide(tmp_path):
+    _, sf, lib = _factory(tmp_path)
+    _album(sf, lib, "rome", approved_name="Same Name")
+    _album(sf, lib, "paris", names=("c.jpg",), approved_name="Same Name")
+
+    plan = _service(sf, lib).build_plan()
+    assert "duplicate_target" in _codes(plan) and plan["state"] == "invalid"
+
+
+def test_destinations_differing_only_by_case_collide(tmp_path):
+    _, sf, lib = _factory(tmp_path)
+    _album(sf, lib, "rome", approved_name="Holiday")
+    _album(sf, lib, "paris", names=("c.jpg",), approved_name="holiday")
+
+    plan = _service(sf, lib).build_plan()
+    assert "unicode_collision" in _codes(plan) and plan["state"] == "invalid"
+
+
+def test_destinations_differing_only_by_unicode_normalization_collide(tmp_path):
+    _, sf, lib = _factory(tmp_path)
+    composed = unicodedata.normalize("NFC", "Café")
+    decomposed = unicodedata.normalize("NFD", "Café")
+    assert composed != decomposed
+    _album(sf, lib, "rome", approved_name=composed)
+    _album(sf, lib, "paris", names=("c.jpg",), approved_name=decomposed)
+
+    plan = _service(sf, lib).build_plan()
+    assert "unicode_collision" in _codes(plan) and plan["state"] == "invalid"
+
+
+def test_case_only_rename_is_flagged_but_not_blocking(tmp_path):
+    _, sf, lib = _factory(tmp_path)
+    _album(sf, lib, "rome", approved_name="Rome")
+
+    plan = _service(sf, lib).build_plan()
+    operation = plan["operations"][0]
+    assert operation["case_only"] is True
+    # A case-only rename needs the staged procedure, but it is not a blocker.
+    assert "destination_exists" not in _codes(plan)
+    assert plan["state"] == "validated"
+
+
+def test_symlinked_source_is_refused(tmp_path):
+    _, sf, lib = _factory(tmp_path)
+    outside = tmp_path / "outside"
+    outside.mkdir()
+    (outside / "a.jpg").write_bytes(b"x" * 32)
+    link = lib / "linked"
+    link.symlink_to(outside, target_is_directory=True)
+
+    with sf() as session:
+        session.add(
+            Asset(
+                id=str(uuid.uuid4()),
+                original_path=str(link / "a.jpg"),
+                current_path=str(link / "a.jpg"),
+                discovered_at=NOW,
+                hash_version=1,
+                byte_size=32,
+            )
+        )
+        session.add(
+            AlbumProposal(
+                id=str(uuid.uuid4()),
+                album="linked",
+                proposed_name="Linked",
+                final_name="Linked",
+                status="approved",
+                version=2,
+            )
+        )
+        session.commit()
+
+    plan = _service(sf, lib).build_plan()
+    assert "symlink" in _codes(plan) and plan["state"] == "invalid"
+
+
+def test_paths_outside_every_library_root_are_refused(tmp_path):
+    _, sf, lib = _factory(tmp_path)
+    stray = tmp_path / "elsewhere" / "rome"
+    stray.mkdir(parents=True)
+    (stray / "a.jpg").write_bytes(b"x" * 32)
+    with sf() as session:
+        session.add(
+            Asset(
+                id=str(uuid.uuid4()),
+                original_path=str(stray / "a.jpg"),
+                current_path=str(stray / "a.jpg"),
+                discovered_at=NOW,
+                hash_version=1,
+                byte_size=32,
+            )
+        )
+        session.add(
+            AlbumProposal(
+                id=str(uuid.uuid4()),
+                album="rome",
+                proposed_name="Rome",
+                final_name="Rome",
+                status="approved",
+                version=2,
+            )
+        )
+        session.commit()
+
+    plan = _service(sf, lib).build_plan()
+    assert "root_escape" in _codes(plan) and plan["state"] == "invalid"
+
+
+def test_excluded_source_is_refused(tmp_path):
+    _, sf, lib = _factory(tmp_path)
+    _album(sf, lib, "_IGNORE/private", approved_name="Private")
+
+    plan = _service(sf, lib).build_plan()
+    assert "excluded_path" in _codes(plan) and plan["state"] == "invalid"
+
+
+# ── export ───────────────────────────────────────────────────────────────────
+
+
+def test_export_is_portable_versioned_and_credential_free(tmp_path):
+    _, sf, lib = _factory(tmp_path)
+    _album(sf, lib, "rome")
+    service = _service(sf, lib)
+    plan = service.build_plan()
+
+    exported = service.export(plan["id"])
+    assert exported["schema_version"] == 1
+    assert exported["checksum"] == plan["checksum"]
+    assert len(exported["operations"]) == 1
+    # Portable: round-trips through JSON unchanged.
+    assert json.loads(json.dumps(exported)) == exported
+    # No credentials or secrets anywhere in the document.
+    flat = json.dumps(exported).lower()
+    for secret in ("api_key", "apikey", "password", "token", "secret"):
+        assert secret not in flat
+
+
+def test_export_of_an_unknown_plan_is_refused(tmp_path):
+    _, sf, lib = _factory(tmp_path)
+    with pytest.raises(RenameError):
+        _service(sf, lib).export("nope")
+
+
+# ── properties ───────────────────────────────────────────────────────────────
+
+
+def test_property_a_validated_plan_has_no_duplicate_destinations(tmp_path):
+    """Across randomized name sets, any plan that validates is collision-free."""
+    rng = random.Random(4242)
+    names = ["Rome", "rome", "Café", unicodedata.normalize("NFD", "Café"), "Paris", "2019 Trip"]
+    for trial in range(25):
+        root = tmp_path / f"trial{trial}"
+        root.mkdir()
+        _, sf, lib = _factory(root)
+        chosen = [rng.choice(names) for _ in range(rng.randint(1, 3))]
+        for index, name in enumerate(chosen):
+            _album(sf, lib, f"album{index}", names=(f"{index}.jpg",), approved_name=name)
+
+        plan = _service(sf, lib).build_plan()
+        destinations = [op["destination_path"] for op in plan["operations"]]
+        if plan["state"] == "validated":
+            folded = [unicodedata.normalize("NFC", d).casefold() for d in destinations]
+            assert len(destinations) == len(set(destinations))
+            assert len(folded) == len(set(folded)), "validated plan has a normalized collision"
+
+
+def test_property_a_plan_preserves_the_asset_set(tmp_path):
+    """Planning never invents, drops, or duplicates an asset."""
+    rng = random.Random(99)
+    for trial in range(15):
+        root = tmp_path / f"trial{trial}"
+        root.mkdir()
+        _, sf, lib = _factory(root)
+        expected = set()
+        for index in range(rng.randint(1, 3)):
+            count = rng.randint(1, 4)
+            _album(
+                sf,
+                lib,
+                f"album{index}",
+                names=tuple(f"{n}.jpg" for n in range(count)),
+                approved_name=f"Album {index}",
+            )
+        with sf() as session:
+            expected = set(session.scalars(select(Asset.id)))
+
+        plan = _service(sf, lib).build_plan()
+        planned = [aid for op in plan["operations"] for aid in op["asset_ids"]]
+        assert sorted(planned) == sorted(set(planned)), "an asset appears twice in one plan"
+        assert set(planned) == expected
+
+
+def test_checksum_is_stable_and_sensitive():
+    base = [{"sequence": 0, "source_path": "/a", "destination_path": "/b", "asset_ids": ["x", "y"]}]
+    same_order_swapped = [
+        {"sequence": 0, "source_path": "/a", "destination_path": "/b", "asset_ids": ["y", "x"]}
+    ]
+    changed = [
+        {"sequence": 0, "source_path": "/a", "destination_path": "/c", "asset_ids": ["x", "y"]}
+    ]
+    assert plan_checksum(base) == plan_checksum(same_order_swapped)
+    assert plan_checksum(base) != plan_checksum(changed)
+
+
+def test_blocking_codes_are_the_documented_set():
+    # Guards against a new issue code silently becoming non-blocking.
+    assert "cross_filesystem" not in BLOCKING_CODES
+    assert "case_only" not in BLOCKING_CODES
+    assert {"source_missing", "destination_exists", "root_escape", "symlink"} <= BLOCKING_CODES
+
+
+# ── API ──────────────────────────────────────────────────────────────────────
+
+
+def test_api_build_get_list_and_export(tmp_path):
+    config, sf, lib = _factory(tmp_path)
+    _album(sf, lib, "rome")
+    before = _snapshot(lib)
+
+    with TestClient(create_app(config)) as client:
+        built = client.post("/api/v1/rename-plans").json()
+        assert built["state"] == "validated"
+        plan_id = built["id"]
+
+        fetched = client.get(f"/api/v1/rename-plans/{plan_id}").json()
+        assert fetched["checksum"] == built["checksum"]
+
+        listed = client.get("/api/v1/rename-plans").json()
+        assert listed["total"] == 1 and listed["items"][0]["id"] == plan_id
+
+        exported = client.get(f"/api/v1/rename-plans/{plan_id}/export").json()
+        assert exported["plan_id"] == plan_id and exported["schema_version"] == 1
+
+        assert client.get("/api/v1/rename-plans/nope").status_code == 404
+
+    assert _snapshot(lib) == before, "the plan API must not touch the filesystem"
+
+
+def test_api_refuses_to_plan_without_approved_proposals(tmp_path):
+    config, sf, lib = _factory(tmp_path)
+    with TestClient(create_app(config)) as client:
+        response = client.post("/api/v1/rename-plans")
+        assert response.status_code == 422
+        assert response.json()["error"]["code"] == "nothing_to_plan"
+
+
+def test_plan_survives_restart(tmp_path):
+    config, sf, lib = _factory(tmp_path)
+    _album(sf, lib, "rome")
+    with TestClient(create_app(config)) as client:
+        plan_id = client.post("/api/v1/rename-plans").json()["id"]
+
+    with TestClient(create_app(config)) as client:
+        after = client.get(f"/api/v1/rename-plans/{plan_id}").json()
+        assert after["state"] == "validated" and after["operation_count"] == 1

tests/story_traceability.json

@@ -84,6 +84,9 @@
     ],
     "US03-05": [
       "tests/e2e/test_phase_c_pipeline.py"
+    ],
+    "US04-01": [
+      "tests/integration/test_rename_plans.py"
     ]
   }
 }