feat: placement reconciler tasks + API (4246, slice 3b)

Three Celery tasks wrapping the 3a service, and the endpoints that drive
them. Routed to `maintenance_long` alongside backups: 33k renames on NFS have
no business in the quick lane where the self-healing sweeps live (the
2026-06-07 starvation).

A durability bug in 3a, found by thinking about what a crash costs rather
than by a failing test: `apply_run` wrote its ledger only at the end, so a
worker dying at row 30,000 of 33,789 would have taken the undo information
for the first 29,999 with it — and that ledger is the ONLY record of where
those files came from. It now persists every 200 moves. Two things fell out
of writing that:

- `_persist` reassigns `run.moves`, so the loop had to snapshot the plan
  first rather than iterate the attribute it rewrites.
- the reassignment is itself load-bearing: SQLAlchemy does not track in-place
  mutation of a JSONB list, so an `.append()` alone would never reach the
  database and the ledger would have stayed silently empty.

Re-running a partially-applied plan is safe — the moved rows no longer match
their `from` and refuse as "row moved since planning" — but `apply_placement`
deliberately has NO autoretry: re-entering a half-applied plan should be the
operator's call after reading what happened, not the queue's.

Endpoints gate on run state as well as the service does, so a stray POST
cannot re-apply an applied run. The list response omits `moves` (an applied
whole-library run carries tens of thousands of entries); the detail endpoint
includes them, because that detail IS the preview read before agreeing.

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

backend/app/api/cleanup.py

@@ -29,7 +29,7 @@ from quart import Blueprint, jsonify, request
 from sqlalchemy import select
 
 from ..extensions import get_session
-from ..models import LibraryAuditRun
+from ..models import LibraryAuditRun, LibraryPlacementRun
 from ..services import cleanup_service, library_layout
 from ._responses import error_response as _bad
 
@@ -218,3 +218,91 @@ async def layout_survey():
             )
         )
     return jsonify({**report.as_dict(), "checked_disk": check_disk})
+
+
+def _serialize_placement_run(run: LibraryPlacementRun, *, moves: bool = False) -> dict:
+    """`moves` is opt-in: an applied whole-library run carries tens of
+    thousands of entries, which is a fine thing to hold in Postgres and a
+    poor thing to put in every list response."""
+    out = {
+        "id": run.id,
+        "status": run.status,
+        "artist_id": run.artist_id,
+        "started_at": run.started_at.isoformat() if run.started_at else None,
+        "finished_at": run.finished_at.isoformat() if run.finished_at else None,
+        "planned_count": run.planned_count,
+        "moved_count": run.moved_count,
+        "refused_count": run.refused_count,
+        "refusals": run.refusals or [],
+        "error": run.error,
+    }
+    if moves:
+        out["moves"] = run.moves or []
+    return out
+
+
+@cleanup_bp.route("/placement/runs", methods=["GET"])
+async def placement_runs():
+    """Newest first. Without `moves`, so the list stays small."""
+    try:
+        limit = min(int(request.args.get("limit", "25")), 100)
+    except ValueError:
+        return _bad("invalid_limit")
+    async with get_session() as session:
+        rows = (await session.execute(
+            select(LibraryPlacementRun)
+            .order_by(LibraryPlacementRun.id.desc()).limit(limit)
+        )).scalars().all()
+    return jsonify({"runs": [_serialize_placement_run(r) for r in rows]})
+
+
+@cleanup_bp.route("/placement/runs/<int:run_id>", methods=["GET"])
+async def placement_run(run_id: int):
+    """One run WITH its moves — this is the preview the operator reads before
+    agreeing, and the record of what happened afterwards."""
+    async with get_session() as session:
+        run = await session.get(LibraryPlacementRun, run_id)
+        if run is None:
+            return _bad("not_found", status=404)
+        return jsonify(_serialize_placement_run(run, moves=True))
+
+
+@cleanup_bp.route("/placement/plan", methods=["POST"])
+async def placement_plan():
+    """Queue a planning run. `artist_id` scopes it to one artist, which is the
+    intended use: do one, look at the gallery, then continue or revert."""
+    body = await request.get_json(silent=True) or {}
+    artist_id = body.get("artist_id")
+    if artist_id is not None and not isinstance(artist_id, int):
+        return _bad("invalid_artist_id")
+    from ..tasks.library_placement import plan_placement
+    plan_placement.delay(artist_id)
+    return jsonify({"status": "dispatched"}), 202
+
+
+@cleanup_bp.route("/placement/runs/<int:run_id>/apply", methods=["POST"])
+async def placement_apply(run_id: int):
+    """Execute a ready run. This renames files and rewrites rows."""
+    async with get_session() as session:
+        run = await session.get(LibraryPlacementRun, run_id)
+        if run is None:
+            return _bad("not_found", status=404)
+        if run.status != "ready":
+            return _bad("not_ready", detail=f"run is {run.status}")
+    from ..tasks.library_placement import apply_placement
+    apply_placement.delay(run_id)
+    return jsonify({"status": "dispatched"}), 202
+
+
+@cleanup_bp.route("/placement/runs/<int:run_id>/revert", methods=["POST"])
+async def placement_revert(run_id: int):
+    """Put an applied run's files back. The reason the ledger is kept."""
+    async with get_session() as session:
+        run = await session.get(LibraryPlacementRun, run_id)
+        if run is None:
+            return _bad("not_found", status=404)
+        if run.status != "applied":
+            return _bad("not_applied", detail=f"run is {run.status}")
+    from ..tasks.library_placement import revert_placement
+    revert_placement.delay(run_id)
+    return jsonify({"status": "dispatched"}), 202

backend/app/celery_app.py

@@ -35,6 +35,7 @@ def make_celery() -> Celery:
             "backend.app.tasks.backup",
             "backend.app.tasks.admin",
             "backend.app.tasks.library_audit",
+            "backend.app.tasks.library_placement",
             "backend.app.tasks.translation",
         ],
     )
@@ -62,6 +63,8 @@ def make_celery() -> Celery:
             # 2026-06-07: a 2h audit blocked vacuum/backup/normalize for hours).
             "backend.app.tasks.maintenance.*": {"queue": "maintenance"},
             "backend.app.tasks.backup.*": {"queue": "maintenance_long"},
+            # 33k renames on NFS: long lane, same as backups.
+            "backend.app.tasks.library_placement.*": {"queue": "maintenance_long"},
             "backend.app.tasks.admin.*": {"queue": "maintenance_long"},
             "backend.app.tasks.library_audit.*": {"queue": "maintenance_long"},
             # Translation backfill hits the LLM (~1–6s/item) → the long lane so it

backend/app/services/library_layout.py

@@ -317,7 +317,7 @@ def _move_one(src: Path, dest: Path) -> str | None:
 
 
 def apply_run(
-    session: Session, run: LibraryPlacementRun,
+    session: Session, run: LibraryPlacementRun, *, chunk: int = 0,
 ) -> LibraryPlacementRun:
     """Execute a `ready` run's stored plan: file and row together, per row.
 
@@ -325,13 +325,38 @@ def apply_run(
     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.
+
+    `chunk` commits progress every N moves. Set it for any real run: the
+    ledger is the ONLY record of where a file came from, so a worker that
+    dies at row 30,000 of 33,789 must not take the undo information for the
+    first 29,999 with it. Left at 0 (tests, small runs) everything persists
+    in one go at the end.
+
+    Re-running a partially-applied plan is safe rather than clever: the rows
+    already moved no longer match their `from`, so they refuse as "row moved
+    since planning" instead of being moved twice.
     """
     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:
+
+    def _persist() -> None:
+        # Reassign rather than mutate: SQLAlchemy does not track in-place
+        # changes to a JSONB list, so an .append() alone would never reach
+        # the database and the ledger would silently stay empty.
+        run.moves = list(moved)
+        run.refusals = list(refusals)
+        run.moved_count = len(moved)
+        run.refused_count = len(refusals)
+        session.commit()
+
+    # Snapshot the plan before iterating: `_persist` reassigns `run.moves`,
+    # and iterating the attribute while rewriting it would walk a list that
+    # changes underneath the loop.
+    plan = list(run.moves)
+    for done, move in enumerate(plan, start=1):
         record = session.get(ImageRecord, move["image_id"])
         if record is None:
             refusals.append({"image_id": move["image_id"], "reason": "row gone"})
@@ -349,6 +374,8 @@ def apply_run(
             continue
         record.path = move["to"]
         moved.append(move)
+        if chunk and done % chunk == 0:
+            _persist()
 
     run.moves = moved
     run.refusals = refusals
@@ -372,6 +399,12 @@ def revert_run(
 
     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.
+
+    A revert interrupted half way is resumable by re-running it: the rows
+    already put back no longer sit at `to`, so they refuse rather than move
+    twice. Unlike the apply this needs no chunked persistence — it consumes
+    the ledger rather than producing it, so a crash costs progress, not
+    information.
     """
     if run.status != "applied":
         raise ValueError(f"run {run.id} is {run.status}, not applied")

backend/app/tasks/library_placement.py

@@ -0,0 +1,125 @@
+"""Placement reconciler tasks — plan, apply, revert (milestone #421).
+
+The service (`services.library_layout`) holds the decisions; this module is
+only the async wrapper, matching `tasks.library_audit`: run on the
+maintenance queue, mark the run `error` with a traceback if anything escapes,
+and return a small summary dict so eager-mode tests can assert on it.
+
+Applying is a long run — 33,789 renames on the operator's library at the time
+of writing — so `apply_placement` persists its ledger in chunks rather than
+at the end. That ledger is the only record of where each file came from, and
+a worker that dies two thirds of the way through must not take the undo
+information for the first two thirds with it.
+"""
+
+import logging
+import traceback
+from datetime import UTC, datetime
+from pathlib import Path
+
+from sqlalchemy.exc import DBAPIError, OperationalError
+
+from ..celery_app import celery
+from ..models import LibraryPlacementRun
+from ..services import library_layout
+from ._sync_engine import sync_session_factory as _sync_session_factory
+
+log = logging.getLogger(__name__)
+
+IMAGES_ROOT = Path("/images")
+
+# Commit the ledger every this many moves. Small enough that a crash loses
+# seconds of work, large enough not to make a COMMIT per rename.
+_APPLY_CHUNK = 200
+
+
+def _fail(session, run_id: int, message: str) -> None:
+    run = session.get(LibraryPlacementRun, run_id)
+    if run is not None:
+        run.status = "error"
+        run.error = message
+        run.finished_at = datetime.now(UTC)
+        session.commit()
+
+
+@celery.task(
+    name="backend.app.tasks.library_placement.plan_placement",
+    autoretry_for=(OperationalError, DBAPIError),
+    retry_backoff=5, retry_backoff_max=60, retry_jitter=True, max_retries=3,
+    soft_time_limit=900, time_limit=1000,
+)
+def plan_placement(artist_id: int | None = None) -> dict:
+    """Build a move plan and leave it `ready` for the operator to read.
+
+    Reads rows and stats destinations; moves nothing.
+    """
+    SessionLocal = _sync_session_factory()
+    with SessionLocal() as session:
+        run = library_layout.plan_placement(
+            session, IMAGES_ROOT, artist_id=artist_id,
+        )
+        session.commit()
+        return {
+            "run_id": run.id, "status": run.status,
+            "planned_count": run.planned_count,
+        }
+
+
+@celery.task(
+    name="backend.app.tasks.library_placement.apply_placement",
+    soft_time_limit=7200, time_limit=7500,
+)
+def apply_placement(run_id: int) -> dict:
+    """Execute a `ready` run's stored plan. Renames files and rewrites rows.
+
+    No autoretry: a retry would re-enter a half-applied plan on a schedule
+    nobody asked for. Re-running IS safe (the applied rows refuse as "row
+    moved since planning"), but that should be the operator's decision after
+    reading what happened, not the queue's.
+    """
+    SessionLocal = _sync_session_factory()
+    with SessionLocal() as session:
+        run = session.get(LibraryPlacementRun, run_id)
+        if run is None:
+            return {"run_id": run_id, "status": "missing"}
+        if run.status != "ready":
+            return {"run_id": run_id, "status": run.status, "skipped": True}
+        try:
+            library_layout.apply_run(session, run, chunk=_APPLY_CHUNK)
+            session.commit()
+        except Exception:
+            log.exception("placement apply failed for run %s", run_id)
+            session.rollback()
+            _fail(session, run_id, traceback.format_exc())
+            return {"run_id": run_id, "status": "error"}
+        return {
+            "run_id": run_id, "status": run.status,
+            "moved": run.moved_count, "refused": run.refused_count,
+        }
+
+
+@celery.task(
+    name="backend.app.tasks.library_placement.revert_placement",
+    soft_time_limit=7200, time_limit=7500,
+)
+def revert_placement(run_id: int) -> dict:
+    """Put an applied run's files back where they came from."""
+    SessionLocal = _sync_session_factory()
+    with SessionLocal() as session:
+        run = session.get(LibraryPlacementRun, run_id)
+        if run is None:
+            return {"run_id": run_id, "status": "missing"}
+        if run.status != "applied":
+            return {"run_id": run_id, "status": run.status, "skipped": True}
+        try:
+            library_layout.revert_run(session, run)
+            session.commit()
+        except Exception:
+            log.exception("placement revert failed for run %s", run_id)
+            session.rollback()
+            _fail(session, run_id, traceback.format_exc())
+            return {"run_id": run_id, "status": "error"}
+        return {
+            "run_id": run_id, "status": run.status,
+            "refused": run.refused_count,
+        }

tests/test_api_placement.py

@@ -0,0 +1,163 @@
+"""The placement reconciler's task + API surface (milestone #421, slice 3b).
+
+The move logic itself is covered in tests/test_library_layout.py; this module
+covers the wrapper — that the tasks are registered and routed, that the
+endpoints gate on run state, and that a list response stays small.
+"""
+
+import pytest
+from sqlalchemy import select
+
+from backend.app.celery_app import celery
+from backend.app.models import Artist, LibraryPlacementRun
+
+pytestmark = pytest.mark.integration
+
+_TASKS = (
+    "backend.app.tasks.library_placement.plan_placement",
+    "backend.app.tasks.library_placement.apply_placement",
+    "backend.app.tasks.library_placement.revert_placement",
+)
+
+
+@pytest.mark.parametrize("name", _TASKS)
+def test_placement_tasks_are_registered(name):
+    assert name in celery.tasks
+
+
+def test_placement_runs_on_the_long_maintenance_lane():
+    """33k renames must not sit in the quick lane, which is where the
+    self-healing sweeps live (the 2026-06-07 starvation)."""
+    routes = celery.conf.task_routes
+    assert routes["backend.app.tasks.library_placement.*"] == {
+        "queue": "maintenance_long"
+    }
+
+
+def _run(db, status="ready", moves=None, artist_id=None):
+    """Adds the row; the caller awaits the flush. `db` is the ASYNC session,
+    so flushing here would leave an un-awaited coroutine and the row would
+    never reach the database."""
+    run = LibraryPlacementRun(
+        status=status, artist_id=artist_id, moves=moves or [],
+        planned_count=len(moves or []),
+    )
+    db.add(run)
+    return run
+
+
+@pytest.mark.asyncio
+async def test_runs_list_omits_the_moves(client, db):
+    """An applied whole-library run carries tens of thousands of entries.
+    Fine in Postgres, wrong in every list response."""
+    run = LibraryPlacementRun(
+        status="applied",
+        moves=[{"image_id": 1, "from": "/images/A/x.png", "to": "/images/a/x.png"}],
+        planned_count=1, moved_count=1,
+    )
+    db.add(run)
+    await db.flush()
+
+    resp = await client.get("/api/cleanup/placement/runs")
+    assert resp.status_code == 200
+    body = await resp.get_json()
+    assert body["runs"][0]["planned_count"] == 1
+    assert "moves" not in body["runs"][0]
+
+
+@pytest.mark.asyncio
+async def test_run_detail_carries_the_moves(client, db):
+    """The detail IS the preview the operator reads before agreeing."""
+    run = LibraryPlacementRun(
+        status="ready",
+        moves=[{"image_id": 7, "from": "/images/Conto/x.png", "to": "/images/conto/x.png"}],
+        planned_count=1,
+    )
+    db.add(run)
+    await db.flush()
+
+    resp = await client.get(f"/api/cleanup/placement/runs/{run.id}")
+    assert resp.status_code == 200
+    body = await resp.get_json()
+    assert body["moves"][0]["from"] == "/images/Conto/x.png"
+    assert body["moves"][0]["to"] == "/images/conto/x.png"
+
+
+@pytest.mark.asyncio
+async def test_run_detail_404s_for_an_unknown_run(client):
+    resp = await client.get("/api/cleanup/placement/runs/999999")
+    assert resp.status_code == 404
+
+
+@pytest.mark.asyncio
+async def test_plan_rejects_a_non_integer_artist(client):
+    resp = await client.post(
+        "/api/cleanup/placement/plan", json={"artist_id": "conto"},
+    )
+    assert resp.status_code == 400
+    assert (await resp.get_json())["error"] == "invalid_artist_id"
+
+
+@pytest.mark.asyncio
+async def test_plan_accepts_an_artist_scope(client, db, monkeypatch):
+    sent = {}
+    from backend.app.tasks import library_placement
+
+    monkeypatch.setattr(
+        library_placement.plan_placement, "delay",
+        lambda artist_id=None: sent.update(artist_id=artist_id),
+    )
+    artist = Artist(name="Conto", slug="conto")
+    db.add(artist)
+    await db.flush()
+
+    resp = await client.post(
+        "/api/cleanup/placement/plan", json={"artist_id": artist.id},
+    )
+    assert resp.status_code == 202
+    assert sent["artist_id"] == artist.id
+
+
+@pytest.mark.asyncio
+async def test_apply_refuses_a_run_that_is_not_ready(client, db):
+    """The gate is here as well as in the service — an applied run must not
+    be re-applied by a stray POST."""
+    run = _run(db, status="applied")
+    await db.flush()
+
+    resp = await client.post(f"/api/cleanup/placement/runs/{run.id}/apply")
+    assert resp.status_code == 400
+    assert (await resp.get_json())["error"] == "not_ready"
+
+
+@pytest.mark.asyncio
+async def test_revert_refuses_a_run_that_was_never_applied(client, db):
+    run = _run(db, status="ready")
+    await db.flush()
+
+    resp = await client.post(f"/api/cleanup/placement/runs/{run.id}/revert")
+    assert resp.status_code == 400
+    assert (await resp.get_json())["error"] == "not_applied"
+
+
+@pytest.mark.asyncio
+async def test_apply_dispatches_for_a_ready_run(client, db, monkeypatch):
+    sent = {}
+    from backend.app.tasks import library_placement
+
+    monkeypatch.setattr(
+        library_placement.apply_placement, "delay",
+        lambda run_id: sent.update(run_id=run_id),
+    )
+    run = _run(db, status="ready")
+    await db.flush()
+
+    resp = await client.post(f"/api/cleanup/placement/runs/{run.id}/apply")
+    assert resp.status_code == 202
+    assert sent["run_id"] == run.id
+    # Dispatch only — the endpoint must not have moved anything itself.
+    still = (await db.execute(
+        select(LibraryPlacementRun.status)
+        .where(LibraryPlacementRun.id == run.id)
+    )).scalar_one()
+    assert still == "ready"