feat: placement reconciler — plan, apply, revert (4246, slice 3a)

Milestone #421 step 3, reframed on the operator's steer: not a one-off
migration but the system that keeps the tree true. The 33,789 misplaced rows
the survey found are just its first run.

The placement half was already done, verified by reading each writer rather
than assuming: downloads have always written `<root>/<slug>/<platform>/`
(gallery_dl.py:523), attach_in_place leaves files where the downloader put
them, and `_copy_to_library` / `_supersede` became canonical in #4244. So
nothing is written off-canon today; what remains is the backlog and a standing
check for future drift.

`LibraryPlacementRun` (migration 0099) holds the plan as JSONB, and that one
structure does three jobs: it is the PREVIEW the operator reads, the list the
APPLY executes (rather than re-deriving the set, so the two cannot disagree),
and — because `from` is retained — the UNDO.

The undo is the point. It makes a 33,789-file operation something to do one
artist at a time, look at in the gallery, and reverse if it reads wrong. That
settles whether artist_id or the folder held the truth (spike #4257) by doing
rather than by arguing it from a 50-row sample.

An applied run is therefore HISTORY, not state — lesson #4226's trap, since
it is the only record of where those files used to be. The model and the
migration both say so: any future retention here may prune ready/cancelled/
error runs, never an applied one.

Everything fails closed. The apply re-checks each row against what the plan
recorded — source still there, destination still free, row still pointing
where the plan said — because a download or a supersede can land in between.
A refusal is recorded with its reason and the run continues; one stale row is
not a reason to abandon the other 33,788. The row is updated only after its
rename lands, so a failed move can never leave `path` naming a file that is
not there.

Writing the collision test caught the code disagreeing with its own comment:
it claimed the first of two rows wanting one destination and skipped the
second, silently picking a winner by iteration order. Now it counts first and
filters after, so genuinely neither is planned.

Thumbnails are sha-addressed, not path-keyed, so they do not move — pinned by
a test.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01LVjrnpQjRgHdvq95rASoiR

backend/app/services/library_layout.py

@@ -16,17 +16,23 @@ writing their own — the house shape for rule 93, snippet #3087. A preview
 that computes its set differently from the apply is a preview that can lie,
 and here the apply RENAMES the operator's art.
 
-Nothing in this module writes. It reads rows, and it stats files when asked.
+## What writes and what does not
+
+`survey_layout` and `plan_placement` are read-only — they report and they
+record a plan. `apply_run` and `revert_run` are the only functions here that
+rename a file or rewrite a row, and each does both for one row at a time,
+updating the row only after its rename lands.
 """
 from __future__ import annotations
 
 from dataclasses import dataclass, field
+from datetime import UTC, datetime
 from pathlib import Path
 
 from sqlalchemy import func, select
 from sqlalchemy.orm import Session
 
-from ..models import Artist, ImageRecord
+from ..models import Artist, ImageRecord, LibraryPlacementRun
 from ..utils.paths import canonical_subdir
 
 # Top-level directories under the images root that are STORES, not artists.
@@ -229,3 +235,167 @@ def survey_layout(
         report.unmovable += layout.unmovable
 
     return report
+
+
+# --- the reconciler: plan -> apply -> revert (#4246) -------------------------
+#
+# The three verbs share one materialised plan rather than each deriving its
+# own set. `plan_placement` writes `LibraryPlacementRun.moves`; `apply_run`
+# executes THAT list; `revert_run` walks it backwards. A preview that can
+# disagree with its apply is the failure this shape exists to prevent, and
+# here the apply renames the operator's art.
+#
+# Every step fails CLOSED. The world moves between planning and applying —
+# a download lands, a supersede rewrites a path, a file is deleted — so the
+# apply re-checks each row against what the plan recorded and declines the
+# ones that moved on, instead of trusting a plan that may be minutes old.
+
+
+def plan_placement(
+    session: Session, images_root: Path, *, artist_id: int | None = None,
+) -> LibraryPlacementRun:
+    """Build (and persist) the move plan. Touches no files.
+
+    `artist_id` scopes the run to one artist, which is how this is meant to be
+    used: do one, look at the gallery, then continue or revert. None plans the
+    whole library.
+    """
+    stmt = select(Artist).order_by(Artist.slug)
+    if artist_id is not None:
+        stmt = stmt.where(Artist.id == artist_id)
+    artists = session.execute(stmt).scalars().all()
+
+    candidates: list[dict] = []
+    wanted: dict[str, int] = {}
+    for artist in artists:
+        rows = session.execute(
+            select(ImageRecord.id, ImageRecord.path)
+            .where(*_misplaced_conditions(images_root, artist.id, artist.slug))
+        ).all()
+        for row_id, path in rows:
+            dest = destination_for(path, images_root, artist.slug)
+            if dest is None or dest.exists():
+                continue
+            key = str(dest)
+            candidates.append({"image_id": row_id, "from": path, "to": key})
+            wanted[key] = wanted.get(key, 0) + 1
+
+    # Two rows wanting one destination: plan NEITHER. Which of them "wins" is
+    # not this sweep's call, and planning one of them would silently pick a
+    # winner by iteration order. Counting first and filtering after is what
+    # makes that true — claiming as we go would quietly keep whichever came
+    # first.
+    moves = [m for m in candidates if wanted[m["to"]] == 1]
+
+    run = LibraryPlacementRun(
+        status="ready", artist_id=artist_id, moves=moves,
+        planned_count=len(moves),
+    )
+    session.add(run)
+    session.flush()
+    return run
+
+
+def _move_one(src: Path, dest: Path) -> str | None:
+    """Rename `src` to `dest`. Returns a refusal reason, or None on success.
+
+    A rename within one filesystem, so no copy and no free space needed. The
+    destination check is not a race-free guarantee — nothing here is — but it
+    turns the common case of "something already landed there" into a refusal
+    instead of a silent overwrite.
+    """
+    if not src.exists():
+        return "source missing"
+    if dest.exists():
+        return "destination occupied"
+    try:
+        dest.parent.mkdir(parents=True, exist_ok=True)
+        src.rename(dest)
+    except OSError as exc:
+        return f"rename failed: {exc}"
+    return None
+
+
+def apply_run(
+    session: Session, run: LibraryPlacementRun,
+) -> LibraryPlacementRun:
+    """Execute a `ready` run's stored plan: file and row together, per row.
+
+    The row is updated ONLY after its rename lands, so a refused or failed
+    move can never leave `ImageRecord.path` pointing at a file that is not
+    there. Refusals are recorded and the run continues — one row that moved
+    on since planning is not a reason to abandon the other 33,788.
+    """
+    if run.status != "ready":
+        raise ValueError(f"run {run.id} is {run.status}, not ready")
+
+    refusals: list[dict] = []
+    moved: list[dict] = []
+    for move in run.moves:
+        record = session.get(ImageRecord, move["image_id"])
+        if record is None:
+            refusals.append({"image_id": move["image_id"], "reason": "row gone"})
+            continue
+        if record.path != move["from"]:
+            # Something rewrote this row since the plan was built — a
+            # supersede, or an earlier run. The plan is stale for it.
+            refusals.append({
+                "image_id": move["image_id"], "reason": "row moved since planning",
+            })
+            continue
+        reason = _move_one(Path(move["from"]), Path(move["to"]))
+        if reason is not None:
+            refusals.append({"image_id": move["image_id"], "reason": reason})
+            continue
+        record.path = move["to"]
+        moved.append(move)
+
+    run.moves = moved
+    run.refusals = refusals
+    run.moved_count = len(moved)
+    run.refused_count = len(refusals)
+    run.status = "applied"
+    run.finished_at = datetime.now(UTC)
+    session.flush()
+    return run
+
+
+def revert_run(
+    session: Session, run: LibraryPlacementRun,
+) -> LibraryPlacementRun:
+    """Put an applied run's files back where they came from.
+
+    This is why `from` is retained. It is the answer to "do one artist, look
+    at it, and undo if it reads wrong" — which is a cheaper way to settle
+    whether artist_id or the folder held the truth (#4257) than arguing it
+    from a sample.
+
+    Refuses the same way the apply does: a file someone has since moved or
+    replaced stays where it is, and its row is left alone.
+    """
+    if run.status != "applied":
+        raise ValueError(f"run {run.id} is {run.status}, not applied")
+
+    refusals: list[dict] = []
+    reverted = 0
+    for move in run.moves:
+        record = session.get(ImageRecord, move["image_id"])
+        if record is None or record.path != move["to"]:
+            refusals.append({
+                "image_id": move["image_id"], "reason": "row changed since apply",
+            })
+            continue
+        reason = _move_one(Path(move["to"]), Path(move["from"]))
+        if reason is not None:
+            refusals.append({"image_id": move["image_id"], "reason": reason})
+            continue
+        record.path = move["from"]
+        reverted += 1
+
+    run.refusals = refusals
+    run.refused_count = len(refusals)
+    run.moved_count = run.moved_count - reverted
+    run.status = "reverted"
+    run.finished_at = datetime.now(UTC)
+    session.flush()
+    return run