US05-02: Orchestrate Album Upload Batches

README.md

@@ -102,3 +102,37 @@ work_item/scripts/python -m pytest tests/e2e -m phase_c -q
 The deterministic naming provider is enabled only by test configuration
 (`PHOTO_PIPELINE_FAKE_NAMING_LOG`); without it the application falls back to the
 offline naming-policy name. Phase A and B suites remain green in the full run above.
+
+### Phase D acceptance gate
+
+Phase D (Epic E04: guarded renaming) is the first phase that changes the library on
+disk, so its gate is the strictest. One command runs the rename API journeys, the
+filesystem fault injection, and the browser suite:
+
+```bash
+work_item/scripts/python -m pytest tests/e2e -m phase_d -q
+```
+
+- `tests/e2e/test_phase_d_pipeline.py` drives a real server over HTTP: plan and
+  export, confirmation with the plan version and checksum (a stale token is refused
+  without touching disk), a valid apply, the case-only rename procedure, a collision
+  whose occupant survives, a source that changed after planning, and durability
+  across a full restart.
+- **Fault injection is real.** `PHOTO_PIPELINE_FAULT_AFTER=<journal state>` kills the
+  server process the instant that state is persisted. The suite crashes it at every
+  journal transition in turn (`moving`, `moved`, `database_updated`, `verified`),
+  starts a fresh process against the same database and library, and requires recovery
+  to converge from journal and disk evidence alone — with the asset set, the stable
+  IDs, and every content hash unchanged. Ambiguous evidence is never guessed: it stays
+  classified `manual` and keeps blocking. An unresolved rename is the cancellation
+  boundary — there is no cancel once a run starts, and unrelated mutations (album
+  proposal generation and approval) are refused with 409 `rename_recovery_required`
+  until it is resolved, while reads stay available.
+- `tests/e2e/test_renames_ui.py` covers the browser journeys: preview of every
+  affected path, confirmation carrying the server-issued token, apply with progress
+  and terminal verification, stale confirmation, collision, interruption, recovery,
+  rollback, keyboard confirmation, and the view still matching the journal after a
+  server restart.
+
+The fault barrier is test-only configuration; without `PHOTO_PIPELINE_FAULT_AFTER`
+the apply path has no crash points. Phases A–C remain green in the full run above.

frontend/css/app.css

@@ -165,6 +165,11 @@ a.link:hover { text-decoration: underline; }
 .badge.approved { color: var(--ok); border-color: var(--ok); }
 .badge.error { color: var(--danger); border-color: var(--danger); }
 
+/* Renames: the preview table must show whole paths, so let them wrap rather than
+   truncate — a hidden path segment is exactly what makes a move a surprise. */
+.grid th, .grid td { text-align: left; padding: 6px 8px; border-bottom: 1px solid var(--border); vertical-align: top; }
+.path { font-family: ui-monospace, monospace; font-size: 0.85em; word-break: break-all; }
+
 /* Narrow screens: stack the panes so blockers and actions stay reachable. */
 @media (max-width: 720px) {
   .two-pane { grid-template-columns: 1fr; }

frontend/index.html

@@ -17,6 +17,7 @@
         <a href="#/library" data-nav="library">Library</a>
         <a href="#/analyze" data-nav="analyze">Analyze</a>
         <a href="#/albums" data-nav="albums">Albums</a>
+        <a href="#/renames" data-nav="renames">Renames</a>
         <a href="#/stats" data-nav="stats">Stats</a>
       </nav>
     </header>

frontend/js/api.js

@@ -99,4 +99,19 @@ export const api = {
       body: JSON.stringify(payload),
       ...opts,
     }),
+
+  // ── Renames: plan, confirm, apply, recover ──────────────────────────────
+  listPlans: (opts = {}) => request("/rename-plans", opts),
+  buildPlan: (opts = {}) => request("/rename-plans", { method: "POST", ...opts }),
+  getPlan: (id, opts = {}) => request(`/rename-plans/${encodeURIComponent(id)}`, opts),
+  applyPlan: (id, payload, opts = {}) =>
+    request(`/rename-plans/${encodeURIComponent(id)}/apply`, {
+      method: "POST",
+      body: JSON.stringify(payload),
+      ...opts,
+    }),
+  rollbackPlan: (id, opts = {}) =>
+    request(`/rename-plans/${encodeURIComponent(id)}/rollback`, { method: "POST", ...opts }),
+  renameRecovery: (opts = {}) => request("/rename-recovery", opts),
+  resolveRecovery: (opts = {}) => request("/rename-recovery/resolve", { method: "POST", ...opts }),
 };

frontend/js/app.js

@@ -1,5 +1,6 @@
 import { api } from "./api.js";
 import { navigate, onRouteChange, parseHash } from "./router.js";
+import { renderRenames, setRenamesRender } from "./renames.js";
 import {
   renderAlbums,
   renderAnalyze,
@@ -362,11 +363,13 @@ function render() {
   else if (path === "/library") renderLibrary(root, params);
   else if (path === "/analyze") renderAnalyze(root, params);
   else if (path === "/albums") renderAlbums(root, params);
+  else if (path === "/renames") renderRenames(root, params);
   else if (path === "/stats") renderStats(root, params);
   else show(errorBanner("Unknown view"));
 }
 
 // Let views re-render the current route after a mutation.
 setRender(render);
+setRenamesRender(render);
 onRouteChange(render);
 render();

frontend/js/renames.js

@@ -0,0 +1,357 @@
+// Renames view (US04-05): preview a rename plan, confirm it with the server-issued
+// token, watch it run, and resolve anything an interruption left behind.
+//
+// Nothing here decides what is safe. Blocking issues, plan applicability, and
+// recovery classifications all come from the server; the view only shows them and
+// refuses to offer an action the server would reject.
+import { api } from "./api.js";
+import { el, errorBanner, setActiveNav } from "./dom.js";
+
+// What the last confirm did, for this tab only. Deliberately not persisted: after a
+// reload the page must show what the journal says, not what this page remembers.
+let outcome = null;
+
+let render = () => {};
+export function setRenamesRender(fn) {
+  render = fn;
+}
+
+const BLOCKED_TITLE = "an unresolved rename must be resolved first";
+
+export async function renderRenames(root, params = {}) {
+  setActiveNav("renames");
+  let plans, recovery;
+  try {
+    [plans, recovery] = await Promise.all([api.listPlans(), api.renameRecovery()]);
+  } catch (error) {
+    root.replaceChildren(errorBanner(`Failed to load renames: ${error.message}`));
+    return;
+  }
+
+  const selectedId = params.plan || (plans.items[0] && plans.items[0].id) || null;
+  let plan = null;
+  if (selectedId) {
+    try {
+      plan = await api.getPlan(selectedId);
+    } catch (error) {
+      root.replaceChildren(errorBanner(`Failed to load plan: ${error.message}`));
+      return;
+    }
+  }
+
+  const blocked = recovery.blocks_mutation;
+  root.replaceChildren(
+    el("h1", {}, "Renames"),
+    recoveryPanel(recovery),
+    el(
+      "div",
+      { class: "toolbar" },
+      el(
+        "button",
+        {
+          class: "primary",
+          "data-testid": "build-plan",
+          disabled: blocked ? "disabled" : false,
+          title: blocked ? BLOCKED_TITLE : false,
+          onclick: () => run(() => api.buildPlan(), "Planning…"),
+        },
+        "Build plan from approved names"
+      )
+    ),
+    outcomeBanner(),
+    plan
+      ? planSection(plan, blocked)
+      : el(
+          "p",
+          { class: "muted", "data-testid": "no-plan" },
+          "No rename plan yet — approve album names, then build a plan to preview the moves."
+        )
+  );
+}
+
+// ── preview ──────────────────────────────────────────────────────────────────
+function planSection(plan, blocked) {
+  const blockingIssues = plan.operations.flatMap((operation) =>
+    operation.issues.filter((issue) => issue.blocking)
+  );
+
+  return el(
+    "div",
+    { "data-testid": "plan" },
+    el(
+      "div",
+      { class: "decision-bar" },
+      el("span", { class: `badge ${plan.state}`, "data-testid": "plan-state" }, plan.state),
+      el(
+        "span",
+        { class: "badge", "data-testid": "operation-count" },
+        `${plan.operation_count} folder${plan.operation_count === 1 ? "" : "s"}`
+      ),
+      el(
+        "span",
+        { class: "badge", "data-testid": "affected-assets" },
+        `${plan.operations.reduce((sum, o) => sum + o.asset_count, 0)} photos`
+      ),
+      blockingIssues.length
+        ? el(
+            "span",
+            { class: "badge attention", "data-testid": "blocking-count" },
+            `${blockingIssues.length} blocking issue${blockingIssues.length === 1 ? "" : "s"}`
+          )
+        : null
+    ),
+    plan.blockers.length
+      ? el(
+          "div",
+          { class: "alert", role: "alert", "data-testid": "plan-blockers" },
+          `This plan cannot run: ${plan.blockers.join("; ")}`
+        )
+      : null,
+    operationsTable(plan),
+    confirmBlock(plan, blocked)
+  );
+}
+
+function operationType(operation) {
+  if (operation.case_only) return "case-only rename";
+  if (!operation.same_filesystem) return "cross-filesystem move";
+  return "rename";
+}
+
+function operationsTable(plan) {
+  if (!plan.operations.length) {
+    return el("p", { class: "muted", "data-testid": "no-operations" }, "This plan moves nothing.");
+  }
+  return el(
+    "table",
+    { class: "grid", "data-testid": "operations" },
+    el(
+      "thead",
+      {},
+      el(
+        "tr",
+        {},
+        ...["#", "Album", "From", "To", "Type", "Photos", "Issues", "State"].map((label) =>
+          el("th", { scope: "col" }, label)
+        )
+      )
+    ),
+    el(
+      "tbody",
+      {},
+      ...plan.operations.map((operation) =>
+        el(
+          "tr",
+          { "data-testid": "operation-row", "data-album": operation.album },
+          el("td", {}, String(operation.sequence)),
+          el("td", { "data-testid": "op-album" }, operation.album),
+          // Full paths, not basenames: the whole point of the preview is that no
+          // move is a surprise.
+          el("td", { class: "path", "data-testid": "op-source" }, operation.source_path),
+          el("td", { class: "path", "data-testid": "op-destination" }, operation.destination_path),
+          el("td", { "data-testid": "op-type" }, operationType(operation)),
+          el("td", { "data-testid": "op-count" }, String(operation.asset_count)),
+          el(
+            "td",
+            { "data-testid": "op-issues" },
+            operation.issues.length
+              ? operation.issues.map((issue) =>
+                  el(
+                    "span",
+                    {
+                      class: `badge ${issue.blocking ? "attention" : ""}`,
+                      "data-testid": "op-issue",
+                      "data-code": issue.code,
+                      title: issue.message,
+                    },
+                    `${issue.blocking ? "blocks" : "note"}: ${issue.code}`
+                  )
+                )
+              : el("span", { class: "muted" }, "—")
+          ),
+          el(
+            "td",
+            {},
+            el("span", { "data-testid": "op-state" }, operation.journal_state),
+            operation.verified_at
+              ? el("span", { class: "badge complete", "data-testid": "op-verified" }, "verified")
+              : null,
+            operation.error_code
+              ? el(
+                  "div",
+                  { class: "blocker", "data-testid": "op-error" },
+                  `${operation.error_code}: ${operation.error_message || ""}`
+                )
+              : null
+          )
+        )
+      )
+    )
+  );
+}
+
+// ── confirmation ─────────────────────────────────────────────────────────────
+// Journal states whose content can still be moved back; `complete` is not one of
+// them (photo_pipeline/services/rename_journal.py: complete is terminal).
+const REVERSIBLE = new Set(["moved", "database_updated", "verified", "rollback_required"]);
+
+function rollbackable(plan) {
+  return plan.operations.some((operation) => REVERSIBLE.has(operation.journal_state));
+}
+
+function confirmBlock(plan, blocked) {
+  const applyable = plan.applicable && !blocked;
+  return el(
+    "div",
+    { class: "card", "data-testid": "confirm" },
+    el("h2", {}, "Confirm"),
+    // The token is shown, not just sent: a confirmation the user cannot see is a
+    // confirmation they cannot check against what the preview claimed.
+    el(
+      "p",
+      { class: "muted", "data-testid": "confirm-token" },
+      `Plan ${plan.id} · version ${plan.version} · checksum ${plan.checksum.slice(0, 12)}`
+    ),
+    el(
+      "p",
+      { "data-testid": "cancel-note" },
+      "Once confirmed, the run cannot be cancelled part-way: each folder is moved, " +
+        "verified, and recorded before the next one starts. An interruption is " +
+        "recoverable from the journal — it is never left half-applied."
+    ),
+    el(
+      "div",
+      { class: "toolbar" },
+      el(
+        "button",
+        {
+          class: "primary",
+          "data-testid": "apply-plan",
+          disabled: applyable ? false : "disabled",
+          title: blocked
+            ? BLOCKED_TITLE
+            : plan.applicable
+              ? false
+              : `plan is ${plan.state} — it cannot be applied`,
+          onclick: () =>
+            run(
+              () =>
+                api.applyPlan(plan.id, {
+                  expected_version: plan.version,
+                  expected_checksum: plan.checksum,
+                }),
+              `Applying ${plan.operation_count} rename(s)…`
+            ),
+        },
+        `Apply ${plan.operation_count} rename(s)`
+      ),
+      // Rollback is recovery, not undo: a completed rename is terminal and stays
+      // that way. The button appears only while operations are still reversible.
+      rollbackable(plan)
+        ? el(
+            "button",
+            {
+              "data-testid": "rollback-plan",
+              onclick: () => run(() => api.rollbackPlan(plan.id), "Rolling back…"),
+            },
+            "Roll back the unfinished moves"
+          )
+        : null
+    ),
+    el("p", { role: "status", "aria-live": "polite", "data-testid": "apply-progress" }, "")
+  );
+}
+
+// Runs a mutating call, showing progress while it is in flight and leaving the
+// result — including a stale-confirmation conflict — visible afterwards.
+async function run(action, progressText) {
+  const progress = document.querySelector('[data-testid="apply-progress"]');
+  if (progress) progress.textContent = progressText;
+  try {
+    outcome = { kind: "ok", result: await action() };
+  } catch (error) {
+    outcome = error.status === 409 ? { kind: "conflict" } : { kind: "error", error };
+  }
+  render();
+}
+
+function outcomeBanner() {
+  if (!outcome) return null;
+  if (outcome.kind === "conflict") {
+    return el(
+      "div",
+      { class: "alert", role: "alert", "data-testid": "conflict" },
+      "This plan changed on the server since it was loaded. Nothing was applied; " +
+        "the preview below is the current state — review it and confirm again."
+    );
+  }
+  if (outcome.kind === "error") {
+    return el(
+      "div",
+      { class: "alert", role: "alert", "data-testid": "apply-error" },
+      `Failed: ${outcome.error.message}`
+    );
+  }
+  const result = outcome.result;
+  if (result.applied === undefined) return null; // a build or recovery call
+  return el(
+    "div",
+    { class: "alert", role: "status", "data-testid": "apply-result" },
+    `Applied ${result.applied}, failed ${result.failed}, skipped ${result.skipped}. ` +
+      `The plan is now ${result.state}.`
+  );
+}
+
+// ── recovery ─────────────────────────────────────────────────────────────────
+function recoveryPanel(recovery) {
+  if (!recovery.items.length) return null;
+  const recoverable = recovery.items.filter((item) => item.classification !== "manual");
+  return el(
+    "div",
+    { class: "alert", role: "alert", "data-testid": "recovery-banner" },
+    el(
+      "strong",
+      {},
+      recovery.blocks_mutation
+        ? "A rename was interrupted — other changes are paused"
+        : "A rename is unresolved"
+    ),
+    el(
+      "p",
+      { "data-testid": "recovery-explanation" },
+      recovery.blocks_mutation
+        ? "Some folders on disk may disagree with the database. Planning, applying, and " +
+          "album-name changes stay blocked until this is resolved; browsing is unaffected."
+        : "These operations never reached a terminal state."
+    ),
+    el(
+      "ul",
+      {},
+      ...recovery.items.map((item) =>
+        el(
+          "li",
+          { "data-testid": "recovery-item", "data-classification": item.classification },
+          `${item.album}: ${item.journal_state} — ${item.classification} — ${item.reason} ` +
+            `(${item.source_path} → ${item.destination_path})`
+        )
+      )
+    ),
+    // Only evidence-safe work is offered. Ambiguous operations get no button at
+    // all — an action the server would refuse must not look available.
+    recoverable.length
+      ? el(
+          "button",
+          {
+            "data-testid": "resolve-recovery",
+            onclick: () => run(() => api.resolveRecovery(), "Resolving…"),
+          },
+          `Resolve ${recoverable.length} recoverable rename(s) from evidence`
+        )
+      : el(
+          "p",
+          { "data-testid": "manual-only" },
+          "No automatic action is safe here: source and destination both exist, or neither " +
+            "does. Inspect the paths above and resolve them by hand."
+        )
+  );
+}

frontend/js/views.js

@@ -327,11 +327,15 @@ export async function renderStats(root) {
 // Approving never renames anything — it records the approved name only.
 export async function renderAlbums(root, params = {}) {
   setActiveNav("albums");
-  let evidence, proposals;
+  let evidence, proposals, recovery;
   try {
-    [evidence, proposals] = await Promise.all([
+    [evidence, proposals, recovery] = await Promise.all([
       api.albumEvidence({ limit: 500 }),
       api.albumProposals(),
+      // An unresolved rename means folder paths on disk may disagree with the
+      // database, so the server refuses name changes until it is resolved. Ask up
+      // front rather than offering a button that is guaranteed to fail.
+      api.renameRecovery(),
     ]);
   } catch (error) {
     root.replaceChildren(errorBanner(`Failed to load albums: ${error.message}`));
@@ -370,17 +374,27 @@ export async function renderAlbums(root, params = {}) {
     })
   );
 
+  const renameBlocked = recovery.blocks_mutation;
   const detail = selected
     ? albumDetail(
         root,
         folders.find((folder) => folder.album === selected),
         byAlbum.get(selected),
-        params
+        params,
+        renameBlocked
       )
     : el("p", { class: "muted" }, "No albums with evidence yet.");
 
   root.replaceChildren(
     el("h1", {}, "Albums"),
+    renameBlocked
+      ? el(
+          "div",
+          { class: "alert", role: "alert", "data-testid": "rename-blocked" },
+          "An interrupted rename is unresolved. Name changes are paused until it is " +
+            "resolved on the Renames page; browsing stays available."
+        )
+      : null,
     el(
       "div",
       { class: "toolbar" },
@@ -389,6 +403,8 @@ export async function renderAlbums(root, params = {}) {
         {
           class: "primary",
           "data-testid": "generate-proposals",
+          disabled: renameBlocked ? "disabled" : false,
+          title: renameBlocked ? "an unresolved rename must be resolved first" : false,
           onclick: async () => {
             try {
               await api.generateProposals({});
@@ -405,7 +421,7 @@ export async function renderAlbums(root, params = {}) {
   );
 }
 
-function albumDetail(root, folder, proposal, params) {
+function albumDetail(root, folder, proposal, params, renameBlocked = false) {
   if (!folder) return el("p", { class: "muted" }, "Album not found.");
 
   const evidenceBlock = el(
@@ -473,8 +489,13 @@ function albumDetail(root, folder, proposal, params) {
     {
       class: "primary",
       "data-testid": "approve",
-      disabled: proposal.stale || proposal.status === "error" ? "disabled" : false,
+      disabled:
-      title: proposal.stale ? "evidence changed — regenerate first" : false,
+        renameBlocked || proposal.stale || proposal.status === "error" ? "disabled" : false,
+      title: renameBlocked
+        ? "an unresolved rename must be resolved first"
+        : proposal.stale
+          ? "evidence changed — regenerate first"
+          : false,
       onclick: () => mutate(root, () => api.approveProposal(proposal.album, { expected_version: proposal.version }), params),
     },
     proposal.status === "approved" ? "Approved" : "Approve name"

migrations/versions/0008_upload_batches.py

@@ -0,0 +1,91 @@
+"""Upload batches and their items (US05-02).
+
+Revision ID: 0008_upload_batches
+Revises: 0007_rename_plans
+Create Date: 2026-08-16
+
+One row per approved album folder handed to immich-go, plus the per-asset
+pre-upload hashes that later stories verify the result against.
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = "0008_upload_batches"
+down_revision = "0007_rename_plans"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "upload_batches",
+        sa.Column("id", sa.String(), primary_key=True),
+        sa.Column("album", sa.String(), nullable=False),
+        sa.Column("folder", sa.String(), nullable=False),
+        sa.Column("album_name", sa.String(), nullable=False),
+        # planned | running | cancelling | cancelled | succeeded | failed
+        # | unknown_requires_verification
+        sa.Column("state", sa.String(), nullable=False, server_default="planned"),
+        # Preflight token this batch was approved against; re-checked before every
+        # attempt, so changed bytes or decisions cannot be uploaded silently.
+        sa.Column("preflight_token", sa.String(), nullable=False),
+        sa.Column("allow_partial", sa.Boolean(), nullable=False, server_default="0"),
+        sa.Column("command", sa.String(), nullable=True),  # JSON array, redacted
+        sa.Column("uploader_version", sa.String(), nullable=True),
+        sa.Column("asset_count", sa.Integer(), nullable=False, server_default="0"),
+        sa.Column("attempt_count", 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("report_path", sa.String(), nullable=True),
+        sa.Column("report_bytes", sa.Integer(), nullable=True),
+        sa.Column("report_truncated", sa.Boolean(), nullable=False, server_default="0"),
+        sa.Column("exit_code", sa.Integer(), nullable=True),
+        sa.Column("error_code", sa.String(), nullable=True),
+        sa.Column("error_message", sa.String(), nullable=True),
+        sa.Column("started_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("finished_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_upload_batches_album", "upload_batches", ["album"])
+    op.create_index("ix_upload_batches_state", "upload_batches", ["state"])
+
+    op.create_table(
+        "upload_items",
+        sa.Column(
+            "batch_id",
+            sa.String(),
+            sa.ForeignKey("upload_batches.id", ondelete="CASCADE"),
+            primary_key=True,
+        ),
+        sa.Column("asset_id", sa.String(), primary_key=True),
+        sa.Column("path", sa.String(), nullable=False),
+        # The bytes as they were when the batch was created; SHA-1 is what Immich
+        # uses to recognise a file it already holds.
+        sa.Column("sha256", sa.String(), nullable=True),
+        sa.Column("sha1", sa.String(), nullable=True),
+        sa.Column("state", sa.String(), nullable=False, server_default="pending"),
+        sa.Column(
+            "updated_at",
+            sa.DateTime(timezone=True),
+            nullable=False,
+            server_default=sa.text("CURRENT_TIMESTAMP"),
+        ),
+    )
+
+
+def downgrade() -> None:
+    op.drop_table("upload_items")
+    op.drop_table("upload_batches")

photo_pipeline/__main__.py

@@ -36,7 +36,7 @@ def main(argv: Sequence[str] | None = None) -> int:
 
         run_migrations(config.database_url)
         engine = create_db_engine(config.database_url)
-        Worker(create_session_factory(engine), worker_id=args.id).run_forever()
+        Worker(create_session_factory(engine), worker_id=args.id, config=config).run_forever()
         return 0
 
     import uvicorn

photo_pipeline/api/app.py

@@ -25,6 +25,7 @@ from photo_pipeline.api.routes import (
     renames,
     safety,
     thumbnails,
+    uploads,
     workflow,
 )
 
@@ -33,6 +34,7 @@ import photo_pipeline.jobs.domain_handlers  # noqa: F401
 from photo_pipeline.config import Config
 from photo_pipeline.db import create_db_engine, create_session_factory, run_migrations
 from photo_pipeline.logging import configure_logging
+from photo_pipeline.services.upload_batches import UploadBatchService
 
 FRONTEND_DIR = Path(__file__).resolve().parents[2] / "frontend"
 
@@ -49,6 +51,9 @@ def create_app(config: Config | None = None) -> FastAPI:
         app.state.config = config
         app.state.engine = engine
         app.state.session_factory = create_session_factory(engine)
+        # An upload whose process died left no outcome behind; resolve it now so the
+        # uploader lane is free and the uncertain batch is visible (US05-02).
+        UploadBatchService(app.state.session_factory, config=config).recover()
         try:
             yield
         finally:
@@ -67,6 +72,7 @@ def create_app(config: Config | None = None) -> FastAPI:
     app.include_router(library.router, prefix="/api/v1")
     app.include_router(albums.router, prefix="/api/v1")
     app.include_router(renames.router, prefix="/api/v1")
+    app.include_router(uploads.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/albums.py

@@ -17,6 +17,7 @@ from photo_pipeline.schemas import (
 )
 from photo_pipeline.services.albums import AlbumService
 from photo_pipeline.services.proposals import ConflictError, ProposalError, ProposalService
+from photo_pipeline.services.rename_journal import RenameJournal
 
 router = APIRouter(tags=["albums"])
 
@@ -37,6 +38,22 @@ def _error(status: int, code: str, message: str) -> JSONResponse:
     return JSONResponse(status_code=status, content={"error": {"code": code, "message": message}})
 
 
+def _unresolved_rename(request: Request) -> JSONResponse | None:
+    """Refuse work that feeds renaming while the library may be half-renamed.
+
+    An unresolved rename means some folder paths on disk disagree with the database,
+    so evidence gathered now — and any approval based on it — could describe a
+    location that no longer exists. Reading stays available; only mutation stops.
+    """
+    if RenameJournal(request.app.state.session_factory).blocks_mutation():
+        return _error(
+            409,
+            "rename_recovery_required",
+            "an unresolved rename is in progress; resolve it before changing album names",
+        )
+    return None
+
+
 @router.get("/albums/evidence")
 def evidence(
     request: Request,
@@ -47,8 +64,8 @@ def evidence(
 
 
 @router.post("/albums/proposals")
-def generate(body: GenerateProposalsRequest, request: Request) -> dict:
+def generate(body: GenerateProposalsRequest, request: Request):
-    return _proposals(request).generate(body.albums)
+    return _unresolved_rename(request) or _proposals(request).generate(body.albums)
 
 
 @router.get("/albums/proposals")
@@ -76,6 +93,9 @@ def edit_proposal(album: str, body: EditProposalRequest, request: Request):
 
 @router.post("/albums/proposals/{album:path}/approve")
 def approve_proposal(album: str, body: ApproveProposalRequest, request: Request):
+    blocked = _unresolved_rename(request)
+    if blocked is not None:
+        return blocked
     try:
         return _proposals(request).approve(album, expected_version=body.expected_version)
     except ConflictError as error:

photo_pipeline/api/routes/renames.py

@@ -9,6 +9,12 @@ from __future__ import annotations
 from fastapi import APIRouter, Request
 from fastapi.responses import JSONResponse
 
+from photo_pipeline.schemas import ApplyPlanRequest
+from photo_pipeline.services.rename_apply import (
+    ApplyConflict,
+    ApplyError,
+    RenameApplyService,
+)
 from photo_pipeline.services.renames import RenameError, RenameService
 
 router = APIRouter(tags=["renames"])
@@ -21,6 +27,13 @@ def _service(request: Request) -> RenameService:
     )
 
 
+def _apply_service(request: Request) -> RenameApplyService:
+    return RenameApplyService(
+        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}})
 
@@ -52,3 +65,42 @@ def export_plan(plan_id: str, request: Request):
         return _service(request).export(plan_id)
     except RenameError as error:
         return _error(404, "not_found", str(error))
+
+
+@router.post("/rename-plans/{plan_id}/apply")
+def apply_plan(plan_id: str, body: ApplyPlanRequest, request: Request):
+    """Apply a confirmed plan. The confirmation must carry the current plan version
+    (and may carry the checksum) so stale browser state cannot run a changed plan."""
+    try:
+        return _apply_service(request).apply(
+            plan_id,
+            expected_version=body.expected_version,
+            expected_checksum=body.expected_checksum,
+        )
+    except ApplyConflict as error:
+        return _error(409, "version_conflict", str(error))
+    except ApplyError as error:
+        return _error(422, "cannot_apply", str(error))
+
+
+@router.get("/rename-recovery")
+def recovery_status(request: Request) -> dict:
+    """Every unresolved rename with its evidence-derived classification."""
+    journal = _apply_service(request).journal
+    return {
+        "blocks_mutation": journal.blocks_mutation(),
+        "items": journal.classify_all(),
+    }
+
+
+@router.post("/rename-recovery/resolve")
+def resolve_recovery(request: Request) -> dict:
+    return _apply_service(request).recover()
+
+
+@router.post("/rename-plans/{plan_id}/rollback")
+def rollback_plan(plan_id: str, request: Request):
+    try:
+        return _apply_service(request).rollback_plan(plan_id)
+    except ApplyError as error:
+        return _error(422, "cannot_rollback", str(error))

photo_pipeline/api/routes/uploads.py

@@ -0,0 +1,113 @@
+"""Upload preflight and batch API (US05-01, US05-02).
+
+Preflight is a command, not a resource read: it contacts the Immich server, hashes
+the current bytes, and issues a token. Creating a batch requires that token, and
+starting one enqueues a durable job on the single ``upload`` lane — the API never
+runs the uploader in the request thread.
+"""
+
+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 UPLOAD_BATCH, UPLOAD_LOCK
+from photo_pipeline.services.jobs import JobBlocked, JobService
+from photo_pipeline.services.upload_batches import (
+    BatchConflict,
+    BatchError,
+    UploadBatchService,
+)
+from photo_pipeline.services.uploads import UploadError, UploadService
+
+router = APIRouter(tags=["uploads"])
+
+
+class PreflightRequest(BaseModel):
+    # ``None`` means every album; an explicit list scopes the check.
+    albums: list[str] | None = None
+    # Approving a partial upload is an explicit act, never a default.
+    allow_partial: bool = False
+
+
+class CreateBatchRequest(PreflightRequest):
+    # The token of the preflight the user approved; a stale one is refused.
+    token: str
+
+
+def _service(request: Request) -> UploadService:
+    return UploadService(request.app.state.session_factory, config=request.app.state.config)
+
+
+def _batches(request: Request) -> UploadBatchService:
+    return UploadBatchService(request.app.state.session_factory, config=request.app.state.config)
+
+
+def _error(status: int, code: str, message: str) -> JSONResponse:
+    return JSONResponse(status_code=status, content={"error": {"code": code, "message": message}})
+
+
+@router.post("/upload-preflight")
+def preflight(request: Request, body: PreflightRequest | None = None):
+    body = body or PreflightRequest()
+    try:
+        return _service(request).preflight(body.albums, allow_partial=body.allow_partial)
+    except UploadError as error:
+        return _error(422, "unknown_album", str(error))
+
+
+@router.post("/upload-batches", status_code=201)
+def create_batches(body: CreateBatchRequest, request: Request):
+    """Turn an approved preflight into one durable batch per album."""
+    try:
+        return {"batches": _batches(request).create(
+            body.albums, token=body.token, allow_partial=body.allow_partial
+        )}
+    except UploadError as error:
+        return _error(422, "unknown_album", str(error))
+    except BatchConflict as error:
+        return _error(409, error.code, str(error))
+    except BatchError as error:
+        return _error(422, error.code, str(error))
+
+
+@router.get("/upload-batches")
+def list_batches(request: Request) -> dict:
+    return {"batches": _batches(request).list()}
+
+
+@router.get("/upload-batches/{batch_id}")
+def get_batch(batch_id: str, request: Request):
+    batch = _batches(request).get(batch_id)
+    if batch is None:
+        return _error(404, "not_found", f"unknown upload batch {batch_id}")
+    return batch
+
+
+@router.post("/upload-batches/{batch_id}/start")
+def start_batch(batch_id: str, request: Request):
+    """Queue the batch on the uploader lane. The worker performs the upload."""
+    service = _batches(request)
+    batch = service.get(batch_id)
+    if batch is None:
+        return _error(404, "not_found", f"unknown upload batch {batch_id}")
+    try:
+        job = JobService(request.app.state.session_factory).enqueue(
+            UPLOAD_BATCH,
+            lock=UPLOAD_LOCK,
+            # One queued attempt per batch attempt: a double-clicked start reuses it.
+            idempotency_key=f"upload:{batch_id}:{batch['attempt_count']}",
+            items=[batch_id],
+        )
+    except JobBlocked as error:
+        return _error(409, error.code, str(error))
+    return {"batch_id": batch_id, "job": job}
+
+
+@router.post("/upload-batches/{batch_id}/cancel")
+def cancel_batch(batch_id: str, request: Request):
+    try:
+        return _batches(request).cancel(batch_id)
+    except BatchError as error:
+        return _error(404 if error.code == "not_found" else 409, error.code, str(error))

photo_pipeline/config.py

@@ -37,6 +37,8 @@ class Config(BaseModel):
 
     vision_api_key: SecretStr | None = None
     immich_api_key: SecretStr | None = None
+    immich_server_url: str = ""
+    immich_go_binary: str = "immich-go"
 
     @property
     def database_path(self) -> Path:

photo_pipeline/integrations/immich_go.py

@@ -0,0 +1,226 @@
+"""immich-go adapter: binary discovery, version, and redacted command preview.
+
+Upload is the only stage that needs credentials, so this module is also the single
+place that knows an API key exists. It never returns, logs, or renders the secret:
+:func:`build_command` produces the real argument list for the uploader, and
+:func:`redact` produces the copy that is safe for the API, the browser, and the
+activity log. Both come from the same builder so the preview can never drift from
+the command that would actually run.
+
+Server reachability uses ``/api/server/ping`` through stdlib ``urllib`` — the app
+has no HTTP client dependency and this is one request.
+
+:func:`run_upload` is the only place the uploader is actually executed. It never
+uses a shell (the argument list goes straight to ``execve``, so no path or album
+name can be interpreted), streams the report to a file with a byte cap so a chatty
+or looping uploader cannot fill the disk, scrubs the API key out of anything the
+process echoes back, and polls a cancellation callback so a running upload can be
+stopped without killing the worker.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import shutil
+import signal
+import subprocess
+import threading
+import time
+import urllib.error
+import urllib.request
+from collections.abc import Callable
+from pathlib import Path
+
+REDACTED = "***"
+PING_PATH = "/api/server/ping"
+PING_TIMEOUT_SECONDS = 5.0
+# Reports are kept in full up to this size; beyond it the tail is dropped and the
+# result is flagged truncated rather than growing without bound (concept §17).
+MAX_REPORT_BYTES = 4_000_000
+UPLOAD_TIMEOUT_SECONDS = 6 * 60 * 60
+POLL_SECONDS = 0.05
+# Grace period between asking the uploader to stop and killing it.
+TERMINATE_GRACE_SECONDS = 10.0
+# How long to wait for the last output after the process is gone. A child that
+# outlived its parent can still hold the pipe; the report is not worth hanging for.
+DRAIN_SECONDS = 2.0
+
+
+def find_binary(binary: str = "immich-go") -> str | None:
+    """Absolute path of the uploader, or ``None`` when it is not installed."""
+    return shutil.which(binary)
+
+
+def version(binary: str = "immich-go") -> str | None:
+    """Reported uploader version, or ``None`` when it is missing or unusable.
+
+    The version is persisted with every batch (concept §8) because immich-go's
+    flags and report text change between releases.
+    """
+    path = find_binary(binary)
+    if path is None:
+        return None
+    try:
+        result = subprocess.run([path, "--version"], capture_output=True, text=True, timeout=30)
+    except (OSError, subprocess.SubprocessError):
+        return None
+    output = (result.stdout or result.stderr or "").strip()
+    return output.splitlines()[0].strip() if output else None
+
+
+def ping(server_url: str, *, timeout: float = PING_TIMEOUT_SECONDS) -> tuple[bool, str | None]:
+    """``(reachable, detail)`` for the configured Immich server.
+
+    A reachable Immich answers ``{"res": "pong"}``. Anything else — wrong host, no
+    Immich, HTTP error — is a blocker with a short human detail. The detail never
+    carries the URL's credentials because the key travels in a header, not the URL.
+    """
+    if not server_url:
+        return False, "no server URL configured"
+    url = server_url.rstrip("/") + PING_PATH
+    try:
+        with urllib.request.urlopen(url, timeout=timeout) as response:  # noqa: S310
+            payload = json.loads(response.read().decode("utf-8") or "{}")
+    except (urllib.error.URLError, OSError, ValueError, TimeoutError) as error:
+        return False, f"{type(error).__name__}: {error}"
+    if payload.get("res") == "pong":
+        return True, None
+    return False, "server did not answer with pong"
+
+
+def build_command(
+    *,
+    binary: str,
+    server_url: str,
+    api_key: str,
+    album_name: str,
+    folder: Path | str,
+) -> list[str]:
+    """The exact upload invocation for one folder-as-album batch."""
+    return [
+        binary,
+        "upload",
+        "from-folder",
+        f"--server={server_url}",
+        f"--api-key={api_key}",
+        f"--album-name={album_name}",
+        str(folder),
+    ]
+
+
+def redact(command: list[str]) -> list[str]:
+    """The same command with every secret-bearing argument masked."""
+    return [f"--api-key={REDACTED}" if arg.startswith("--api-key=") else arg for arg in command]
+
+
+def run_upload(
+    command: list[str],
+    *,
+    report_path: Path | str,
+    secret: str | None = None,
+    max_report_bytes: int | None = None,  # resolved at call time; see MAX_REPORT_BYTES
+    cancelled: Callable[[], bool] | None = None,
+    timeout: float = UPLOAD_TIMEOUT_SECONDS,
+) -> dict:
+    """Run one upload and return its outcome.
+
+    ``{"exit_code", "cancelled", "timed_out", "report_path", "report_bytes",
+    "report_truncated"}``. Output is streamed to ``report_path`` with ``secret``
+    masked and the file capped at ``max_report_bytes``; the pipe keeps being drained
+    after the cap so the child never blocks on a full buffer. ``cancelled`` is polled
+    while the process runs: when it returns true the uploader is asked to stop, then
+    killed if it does not.
+    """
+    report_path = Path(report_path)
+    report_path.parent.mkdir(parents=True, exist_ok=True)
+    max_report_bytes = MAX_REPORT_BYTES if max_report_bytes is None else max_report_bytes
+    needle = (secret or "").encode() or None
+    written = 0
+    truncated = False
+
+    process = subprocess.Popen(  # noqa: S603 — argv list, never a shell string
+        command,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.STDOUT,
+        shell=False,
+        # Own process group: stopping the upload must stop whatever the uploader
+        # spawned too, not leave orphans holding the pipe open.
+        start_new_session=True,
+    )
+
+    def _drain() -> None:
+        nonlocal written, truncated
+        with open(report_path, "wb") as report:
+            for line in process.stdout:  # line granularity keeps the mask reliable
+                if needle:
+                    line = line.replace(needle, REDACTED.encode())
+                if written >= max_report_bytes:
+                    truncated = True
+                    continue  # keep draining the pipe, stop growing the file
+                room = max_report_bytes - written
+                report.write(line[:room])
+                written += min(len(line), room)
+                truncated = truncated or len(line) > room
+                report.flush()
+
+    reader = threading.Thread(target=_drain, daemon=True)
+    reader.start()
+
+    stopped = timed_out = False
+    deadline = time.monotonic() + timeout
+    while process.poll() is None:
+        if cancelled is not None and cancelled():
+            stopped = True
+        elif time.monotonic() >= deadline:
+            timed_out = True
+        if stopped or timed_out:
+            _stop(process)
+            break
+        time.sleep(POLL_SECONDS)
+
+    exit_code = process.wait()
+    reader.join(timeout=DRAIN_SECONDS)
+    if process.stdout is not None:
+        process.stdout.close()
+    return {
+        "exit_code": exit_code,
+        "cancelled": stopped,
+        "timed_out": timed_out,
+        "report_path": str(report_path),
+        "report_bytes": written,
+        "report_truncated": truncated,
+    }
+
+
+def _stop(process: subprocess.Popen) -> None:
+    """Ask the uploader's whole process group to stop, then kill what remains."""
+    _signal_group(process, signal.SIGTERM)
+    try:
+        process.wait(timeout=TERMINATE_GRACE_SECONDS)
+    except subprocess.TimeoutExpired:
+        _signal_group(process, signal.SIGKILL)
+        process.wait()
+
+
+def _signal_group(process: subprocess.Popen, sig: int) -> None:
+    try:
+        os.killpg(os.getpgid(process.pid), sig)
+    except (ProcessLookupError, PermissionError, OSError):
+        # No group (already reaped, or a platform without them): signal the child.
+        process.send_signal(sig)
+
+
+def preview_command(
+    *, binary: str, server_url: str, album_name: str, folder: Path | str
+) -> list[str]:
+    """Redacted preview built without ever handling the real key."""
+    return redact(
+        build_command(
+            binary=binary,
+            server_url=server_url,
+            api_key=REDACTED,
+            album_name=album_name,
+            folder=folder,
+        )
+    )

photo_pipeline/jobs/domain_handlers.py

@@ -1,10 +1,10 @@
-"""Domain job handlers: safety scoring and content analysis (US02-06).
+"""Domain job handlers: safety scoring, content analysis, uploads (US02-06, US05-02).
 
-Importing this module registers the ``safety_score`` and ``analysis`` job types so
+Importing this module registers the ``safety_score``, ``analysis``, and
-the generic worker can run them per item (one item = one ``asset_id``). Each handler
+``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.
+attempt, and an upload batch refuses to re-run an attempt whose outcome is unknown.
 
 Providers/models are the service defaults here (real NsfwModel / vision provider);
 tests exercise the services directly with injected fakes rather than the worker.
@@ -12,12 +12,15 @@ tests exercise the services directly with injected fakes rather than the worker.
 
 from __future__ import annotations
 
-from photo_pipeline.jobs.handlers import JobContext, register
+from photo_pipeline.jobs.handlers import Cancelled, JobContext, register
 
 SAFETY_SCORE = "safety_score"
 ANALYSIS = "analysis"
+UPLOAD_BATCH = "upload_batch"
 # 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"
 
 
 def _safety_score_item(asset_id: str, ctx: JobContext) -> None:
@@ -32,5 +35,22 @@ def _analysis_item(asset_id: str, ctx: JobContext) -> None:
     AnalysisService(ctx.session_factory).run([asset_id])
 
 
+def _upload_batch_item(batch_id: str, ctx: JobContext) -> None:
+    """One item = one album batch. The upload itself is long and external, so the
+    handler hands the job's cancellation check to the service, which stops the
+    uploader and leaves a resumable batch."""
+    from photo_pipeline.config import Config
+    from photo_pipeline.services.upload_batches import BatchState, UploadBatchService
+
+    config = ctx.config if ctx.config is not None else Config.from_env()
+    service = UploadBatchService(ctx.session_factory, config=config)
+    batch = service.run(batch_id, worker_id=ctx.worker_id, cancelled=ctx.cancelled)
+    if batch["state"] == BatchState.CANCELLED:
+        raise Cancelled(f"upload batch {batch_id} was cancelled")
+    if batch["state"] in (BatchState.FAILED, BatchState.UNKNOWN):
+        raise RuntimeError(f"upload batch {batch_id} is {batch['state']}: {batch['error_code']}")
+
+
 register(SAFETY_SCORE, _safety_score_item)
 register(ANALYSIS, _analysis_item)
+register(UPLOAD_BATCH, _upload_batch_item)

photo_pipeline/jobs/handlers.py

@@ -34,6 +34,9 @@ class JobContext:
     fencing_token: int
     service: "JobService"
     session_factory: object | None = None
+    # Handlers that talk to an external tool (uploads) need the typed configuration;
+    # the worker passes its own so a test stack is never read from the environment.
+    config: object | None = None
 
     def cancelled(self) -> bool:
         from photo_pipeline.services.jobs import JobState

photo_pipeline/jobs/worker.py

@@ -30,6 +30,7 @@ class Worker:
         *,
         job_types: Sequence[str] | None = None,
         lease_seconds: int = 60,
+        config: object | None = None,
     ) -> None:
         self._session_factory = session_factory
         self.service = JobService(session_factory)
@@ -37,6 +38,7 @@ class Worker:
         self.worker_id = worker_id
         self.job_types = list(job_types if job_types is not None else self.handlers.keys())
         self.lease_seconds = lease_seconds
+        self.config = config
 
     def run_once(self) -> str | None:
         """Recover stragglers, then claim and fully process one job. Returns its id."""
@@ -51,7 +53,9 @@ class Worker:
 
     def _process(self, job_id: str, job_type: str, token: int) -> None:
         handler = self.handlers[job_type]
-        ctx = JobContext(job_id, self.worker_id, token, self.service, self._session_factory)
+        ctx = JobContext(
+            job_id, self.worker_id, token, self.service, self._session_factory, self.config
+        )
         self._reset_interrupted_items(job_id, token)
 
         cancelled = False

photo_pipeline/models/__init__.py

@@ -14,6 +14,7 @@ from photo_pipeline.models.duplicates import (
 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.uploads import UploadBatch, UploadItem
 from photo_pipeline.models.workflow import AnalysisResult, SafetyReview
 
 __all__ = [
@@ -29,6 +30,8 @@ __all__ = [
     "RenamePlan",
     "RenameOperation",
     "Thumbnail",
+    "UploadBatch",
+    "UploadItem",
     "SafetyReview",
     "AnalysisResult",
 ]

photo_pipeline/models/uploads.py

@@ -0,0 +1,90 @@
+"""Upload batch persistence (US05-02).
+
+One batch is one approved album folder handed to ``immich-go``. It is the durable
+record of an irreversible external action, so it stores everything needed to answer
+"which exact bytes did we send, with which command, and how did it end?" after a
+crash: the preflight token that authorised it, the redacted command, the uploader
+version, the per-asset pre-upload hashes, every attempt, and where the raw report
+was written.
+
+Item rows keep both digests: SHA-256 is the app's byte identity and SHA-1 is what
+Immich/immich-go use to recognise a file it already has (concept §8).
+"""
+
+from __future__ import annotations
+
+from datetime import datetime
+
+from sqlalchemy import (
+    Boolean,
+    DateTime,
+    ForeignKey,
+    Integer,
+    String,
+    func,
+)
+from sqlalchemy.orm import Mapped, mapped_column
+
+from photo_pipeline.db import Base
+
+
+class UploadBatch(Base):
+    __tablename__ = "upload_batches"
+
+    id: Mapped[str] = mapped_column(String, primary_key=True)
+    album: Mapped[str] = mapped_column(String, nullable=False, index=True)
+    folder: Mapped[str] = mapped_column(String, nullable=False)
+    album_name: Mapped[str] = mapped_column(String, nullable=False)
+    # planned | running | cancelling | cancelled | succeeded | failed
+    # | unknown_requires_verification
+    state: Mapped[str] = mapped_column(String, nullable=False, default="planned")
+    # The preflight token this batch was approved against; re-checked before every
+    # attempt so changed bytes or decisions cannot be uploaded silently.
+    preflight_token: Mapped[str] = mapped_column(String, nullable=False)
+    allow_partial: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
+
+    command: Mapped[str | None] = mapped_column(String)  # JSON array, redacted
+    uploader_version: Mapped[str | None] = mapped_column(String)
+    asset_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
+
+    attempt_count: 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 its outcome.
+    version: Mapped[int] = mapped_column(Integer, nullable=False, default=1)
+    worker_id: Mapped[str | None] = mapped_column(String)
+
+    report_path: Mapped[str | None] = mapped_column(String)
+    report_bytes: Mapped[int | None] = mapped_column(Integer)
+    report_truncated: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
+    exit_code: Mapped[int | None] = mapped_column(Integer)
+    error_code: Mapped[str | None] = mapped_column(String)
+    error_message: Mapped[str | None] = mapped_column(String)
+
+    started_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+    finished_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 UploadItem(Base):
+    __tablename__ = "upload_items"
+
+    batch_id: Mapped[str] = mapped_column(
+        ForeignKey("upload_batches.id", ondelete="CASCADE"), primary_key=True
+    )
+    asset_id: Mapped[str] = mapped_column(String, primary_key=True)
+    path: Mapped[str] = mapped_column(String, nullable=False)
+    # Hashes of the bytes as they were when the batch was created.
+    sha256: Mapped[str | None] = mapped_column(String)
+    sha1: Mapped[str | None] = mapped_column(String)
+    # pending | sent | failed — the per-asset upload *result* is parsed from the
+    # report in US05-03 and verified in US05-04; ``sent`` only means the batch
+    # process exited successfully.
+    state: Mapped[str] = mapped_column(String, nullable=False, default="pending")
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now()
+    )

photo_pipeline/schemas/__init__.py

@@ -6,10 +6,12 @@ from photo_pipeline.schemas.albums import (
     GenerateProposalsRequest,
 )
 from photo_pipeline.schemas.duplicates import DecisionRequest
+from photo_pipeline.schemas.renames import ApplyPlanRequest
 from photo_pipeline.schemas.jobs import JobStartRequest
 from photo_pipeline.schemas.safety import SafetyDecisionRequest
 
 __all__ = [
+    "ApplyPlanRequest",
     "ApproveProposalRequest",
     "DecisionRequest",
     "EditProposalRequest",

photo_pipeline/schemas/renames.py

@@ -0,0 +1,11 @@
+"""Rename apply/confirmation request contracts."""
+
+from __future__ import annotations
+
+from pydantic import BaseModel
+
+
+class ApplyPlanRequest(BaseModel):
+    # The server-issued plan version the user confirmed; a mismatch is a 409.
+    expected_version: int
+    expected_checksum: str | None = None

photo_pipeline/services/hashing.py

@@ -24,7 +24,16 @@ _CHUNK = 1 << 20
 
 
 def sha256_file(path: Path | str) -> str:
-    digest = hashlib.sha256()
+    return _digest_file(path, hashlib.sha256())
+
+
+def sha1_file(path: Path | str) -> str:
+    """SHA-1 of the file bytes. Not an identity hash here — it is the checksum
+    Immich/immich-go use to recognise an asset they already hold (concept §8)."""
+    return _digest_file(path, hashlib.sha1())
+
+
+def _digest_file(path: Path | str, digest) -> str:
     with open(path, "rb") as handle:
         for chunk in iter(lambda: handle.read(_CHUNK), b""):
             digest.update(chunk)

photo_pipeline/services/rename_apply.py

@@ -0,0 +1,493 @@
+"""Apply and recover guarded renames (US04-03, US04-04).
+
+This is the only module that moves a photo library's files. Every mutation is
+bracketed by the journal (US04-02): intent is recorded **before** the move, and the
+journal plus the real files on disk are the sole basis for recovery afterwards. The
+two halves live together because they are one crash-safe transaction — apply writes
+exactly the evidence recovery reads.
+
+Per operation the sequence is:
+
+```
+journal.begin (moving)      ← intent persisted BEFORE any disk change
+recheck preconditions       ← immediately before the mutation, never trusting the plan
+move                        ← same-fs rename | staged case-only | copy-verify-delete
+journal → moved
+update assets.current_path + path occurrences (one transaction)
+journal → database_updated
+verify bytes and paths on disk
+journal → verified → complete
+```
+
+Safety rules that are never relaxed:
+
+- An unexpected occupant at the destination is **never** overwritten; the operation
+  fails and the source is left untouched.
+- Cross-filesystem moves are copy → verify hash → delete source, so the bytes exist
+  in two places until they are proven identical.
+- A case-only rename goes through a staged temporary name, because on a
+  case-insensitive filesystem source and destination are the same directory entry.
+- Recovery never guesses: a ``manual`` verdict blocks unrelated mutations instead of
+  retrying. Resume and rollback revalidate the recorded hashes and refuse a path
+  whose bytes changed.
+- Recovery and rollback are idempotent — replaying them after repeated restarts
+  converges 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.models import Asset, AssetPath, RenamePlan
+from photo_pipeline.services import hashing
+from photo_pipeline.services.rename_journal import (
+    MANUAL,
+    RESUMABLE,
+    JournalState,
+    RenameJournal,
+)
+
+# ``applied`` is included so a repeated apply is a safe no-op rather than an error:
+# every operation is already complete and simply skips (idempotent confirmation).
+# ``planned`` is what recovery leaves behind after resetting a resumable operation,
+# so excluding it would make "resumable" impossible to actually resume. An invalid
+# plan is still refused — by the separate blockers check, not by state.
+APPLYABLE_PLAN_STATES = frozenset({"planned", "validated", "applying", "failed", "applied"})
+
+
+class ApplyError(RuntimeError):
+    """The request cannot be applied (unknown plan, invalid state, blocked)."""
+
+
+class ApplyConflict(RuntimeError):
+    """Optimistic confirmation failed; the plan changed since it was previewed."""
+
+
+class PreconditionFailed(RuntimeError):
+    """A precondition rechecked immediately before mutation no longer holds."""
+
+    def __init__(self, code: str, message: str) -> None:
+        super().__init__(message)
+        self.code = code
+
+
+def _now() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+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.
+    """
+    if os.environ.get("PHOTO_PIPELINE_FAULT_AFTER") == state:
+        os._exit(9)
+
+
+class RenameApplyService:
+    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)
+        self.journal = RenameJournal(session_factory)
+
+    # ── apply ─────────────────────────────────────────────────────────────────
+
+    def apply(
+        self,
+        plan_id: str,
+        *,
+        expected_version: int,
+        expected_checksum: str | None = None,
+        worker_id: str = "apply",
+    ) -> dict:
+        """Apply a confirmed plan. Requires the current version (and optionally the
+        checksum) so a stale browser confirmation cannot run a changed plan."""
+        plan = self._require_plan(plan_id)
+        if plan["version"] != expected_version:
+            raise ApplyConflict(
+                f"plan {plan_id} is at version {plan['version']}, expected {expected_version}"
+            )
+        if expected_checksum is not None and plan["checksum"] != expected_checksum:
+            raise ApplyConflict(f"plan {plan_id} checksum changed since it was previewed")
+        if plan["state"] not in APPLYABLE_PLAN_STATES:
+            raise ApplyError(f"plan {plan_id} is {plan['state']} and cannot be applied")
+        if plan["blockers"]:
+            raise ApplyError(f"plan {plan_id} has unresolved blockers: {plan['blockers']}")
+
+        # Exclusive mutation lease: refuse while any other rename may have the
+        # library half-moved (concept §16 lock hierarchy).
+        blocking = [row for row in self.journal.incomplete() if row["plan_id"] != plan_id]
+        if blocking:
+            raise ApplyError(
+                f"another rename is unresolved ({blocking[0]['id']}); recover it first"
+            )
+
+        token = self._claim_plan(plan_id)
+        applied = failed = skipped = 0
+        for operation in self.journal.operations(plan_id):
+            if operation["journal_state"] == JournalState.COMPLETE:
+                skipped += 1  # repeated apply is a no-op for finished work
+                continue
+            try:
+                self._apply_one(operation, token=token, worker_id=worker_id)
+                applied += 1
+            except PreconditionFailed as error:
+                self.journal.transition(
+                    operation["id"],
+                    JournalState.FAILED,
+                    fencing_token=token,
+                    error=(error.code, str(error)),
+                )
+                failed += 1
+            except Exception as error:  # unexpected: record and stop touching disk
+                self.journal.transition(
+                    operation["id"],
+                    JournalState.FAILED,
+                    fencing_token=token,
+                    error=("apply_error", str(error)),
+                )
+                failed += 1
+        state = self.journal.sync_plan_state(plan_id)
+        return {
+            "plan_id": plan_id,
+            "applied": applied,
+            "failed": failed,
+            "skipped": skipped,
+            "state": state,
+        }
+
+    def _apply_one(self, operation: 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(JournalState.MOVING)
+
+        # 2. Recheck preconditions immediately before mutating, never trusting the
+        #    plan's snapshot: files can change between preview and confirmation.
+        self._recheck(operation, source, destination)
+
+        # 3. The move itself.
+        if operation["case_only"]:
+            self._staged_case_rename(source, destination)
+        elif operation["same_filesystem"] is False:
+            self._copy_verify_delete(source, destination)
+        else:
+            os.rename(source, destination)
+
+        self.journal.transition(operation["id"], JournalState.MOVED, fencing_token=token)
+        _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)
+
+        # 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)
+
+        self.journal.transition(operation["id"], JournalState.COMPLETE, fencing_token=token)
+        _maybe_fault(JournalState.COMPLETE)
+
+    def _recheck(self, operation: dict, source: Path, destination: Path) -> 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 rename through a symlink")
+        # An unexpected occupant is never overwritten. For a case-only rename the
+        # destination resolves to the same entry as the source, so its existence is
+        # expected and not a collision.
+        if destination.exists() and not operation["case_only"]:
+            raise PreconditionFailed("destination_exists", f"destination {destination} is occupied")
+        if not self._within_roots(source) or not self._within_roots(destination):
+            raise PreconditionFailed("root_escape", "path is outside the library roots")
+
+        expected = operation.get("expected_sha256") or {}
+        with self._session_factory() as session:
+            assets = session.scalars(
+                select(Asset).where(Asset.id.in_(operation["asset_ids"]))
+            ).all()
+            current = {asset.id: asset.current_path for asset in assets}
+        for asset_id, path in current.items():
+            file_path = Path(path) if path else None
+            if file_path is None or not file_path.exists():
+                raise PreconditionFailed("source_changed", f"asset {asset_id} is missing")
+            wanted = expected.get(asset_id)
+            if wanted and hashing.sha256_file(file_path) != wanted:
+                raise PreconditionFailed(
+                    "source_changed", f"asset {asset_id} changed since the plan was built"
+                )
+
+    def _within_roots(self, path: Path) -> bool:
+        if not self._roots:
+            return False
+        return any(path == root or root in path.parents for root in self._roots)
+
+    # ── move procedures ───────────────────────────────────────────────────────
+
+    @staticmethod
+    def _staged_case_rename(source: Path, destination: Path) -> None:
+        """Case-only rename via a temporary name: on a case-insensitive filesystem
+        source and destination are the same entry, so a direct rename is a no-op or
+        an error depending on the platform."""
+        staging = source.with_name(f".rename-{uuid.uuid4().hex[:8]}")
+        os.rename(source, staging)
+        try:
+            os.rename(staging, destination)
+        except OSError:
+            os.rename(staging, source)  # put it back rather than leave it staged
+            raise
+
+    @staticmethod
+    def _copy_verify_delete(source: Path, destination: Path) -> None:
+        """Cross-filesystem move: the bytes exist in both places until they are
+        proven identical, and only then is the source removed."""
+        staging = destination.with_name(f".rename-{uuid.uuid4().hex[:8]}")
+        shutil.copytree(source, staging)
+        for original in sorted(p for p in source.rglob("*") if p.is_file()):
+            copied = staging / original.relative_to(source)
+            if not copied.exists():
+                shutil.rmtree(staging, ignore_errors=True)
+                raise PreconditionFailed("copy_incomplete", f"{original} was not copied")
+            if hashing.sha256_file(original) != hashing.sha256_file(copied):
+                shutil.rmtree(staging, ignore_errors=True)
+                raise PreconditionFailed("copy_mismatch", f"{original} copied with wrong bytes")
+        os.rename(staging, destination)
+        shutil.rmtree(source)
+
+    # ── database reconciliation ───────────────────────────────────────────────
+
+    def _reconcile_paths(self, operation: dict, source: Path, destination: Path) -> None:
+        """Re-point every asset in the operation at its new path, keeping the stable
+        ID and recording the path history."""
+        now = _now()
+        with self._session_factory() as session:
+            assets = session.scalars(
+                select(Asset).where(Asset.id.in_(operation["asset_ids"]))
+            ).all()
+            for asset in assets:
+                if not asset.current_path:
+                    continue
+                old = Path(asset.current_path)
+                try:
+                    relative = old.relative_to(source)
+                except ValueError:
+                    continue  # already re-pointed by an earlier partial run
+                new_path = destination / relative
+                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
+                session.add(
+                    AssetPath(
+                        asset_id=asset.id,
+                        path=str(new_path),
+                        valid_from=now,
+                        reason="rename",
+                    )
+                )
+                asset.current_path = str(new_path)
+                asset.state_version += 1
+                asset.updated_at = now
+            session.commit()
+
+    def _verify(self, operation: dict, destination: Path) -> None:
+        """Postcondition: every asset's recorded path exists and its bytes are the
+        ones the plan expected."""
+        expected = operation.get("expected_sha256") or {}
+        with self._session_factory() as session:
+            assets = session.scalars(
+                select(Asset).where(Asset.id.in_(operation["asset_ids"]))
+            ).all()
+            paths = {asset.id: asset.current_path for asset in assets}
+        for asset_id, path in paths.items():
+            if not path or not Path(path).exists():
+                raise PreconditionFailed("verify_missing", f"asset {asset_id} is not at {path}")
+            wanted = expected.get(asset_id)
+            if wanted and hashing.sha256_file(path) != wanted:
+                raise PreconditionFailed("verify_bytes", f"asset {asset_id} bytes changed")
+
+    # ── recovery (US04-04) ────────────────────────────────────────────────────
+
+    def recover(self, *, worker_id: str = "recovery") -> dict:
+        """Resolve every incomplete operation from journal + disk evidence.
+
+        Idempotent: running it repeatedly converges. Ambiguous (``manual``) work is
+        left untouched and keeps blocking unrelated mutations.
+        """
+        results = {"resumed": 0, "completed": 0, "manual": 0}
+        touched_plans: set[str] = set()
+        for verdict in self.journal.classify_all():
+            operation = self.journal.get(verdict["operation_id"])
+            touched_plans.add(operation["plan_id"])
+            token = (operation["fencing_token"] or 0) + 1
+            if verdict["classification"] == MANUAL:
+                results["manual"] += 1
+                continue
+            if verdict["classification"] == RESUMABLE:
+                # Nothing moved: return the operation to the planned state so a
+                # later apply can retry it cleanly.
+                self._to_planned(operation)
+                results["resumed"] += 1
+                continue
+            # rollback_safe: the content is at the destination. Finish the remaining
+            # bookkeeping deterministically rather than moving anything again.
+            self._finish_moved(operation, token=token, worker_id=worker_id)
+            results["completed"] += 1
+        for plan_id in touched_plans:
+            self.journal.sync_plan_state(plan_id)
+        return results
+
+    def _to_planned(self, operation: dict) -> None:
+        state = operation["journal_state"]
+        if state == JournalState.PLANNED:
+            return
+        if state in (JournalState.MOVING, JournalState.FAILED):
+            self.journal.transition(operation["id"], JournalState.PLANNED)
+        elif state == JournalState.ROLLBACK_REQUIRED:
+            self.journal.transition(operation["id"], JournalState.ROLLED_BACK)
+
+    def _finish_moved(self, operation: dict, *, token: int, worker_id: str) -> None:
+        """Drive an operation whose bytes are already at the destination through the
+        remaining states, re-running the idempotent bookkeeping."""
+        operation_id = operation["id"]
+        source = Path(operation["source_path"])
+        destination = Path(operation["destination_path"])
+        state = operation["journal_state"]
+
+        if state == JournalState.ROLLBACK_REQUIRED:
+            self.rollback_operation(operation_id, token=token)
+            return
+
+        if state == JournalState.MOVING:
+            self.journal.transition(operation_id, JournalState.MOVED, fencing_token=token)
+            state = JournalState.MOVED
+        if state == JournalState.MOVED:
+            self._reconcile_paths(operation, source, destination)
+            self.journal.transition(
+                operation_id, JournalState.DATABASE_UPDATED, fencing_token=token
+            )
+            state = JournalState.DATABASE_UPDATED
+        if state == JournalState.DATABASE_UPDATED:
+            self._verify(operation, destination)
+            self.journal.transition(operation_id, JournalState.VERIFIED, fencing_token=token)
+            state = JournalState.VERIFIED
+        if state == JournalState.VERIFIED:
+            self.journal.transition(operation_id, JournalState.COMPLETE, fencing_token=token)
+
+    # ── rollback ──────────────────────────────────────────────────────────────
+
+    def rollback_operation(self, operation_id: str, *, token: int | None = None) -> dict:
+        """Move an operation's content back to its source, revalidating the recorded
+        hashes first. Refuses when the source is occupied or the bytes changed."""
+        operation = self.journal.get(operation_id)
+        if operation is None:
+            raise ApplyError(f"unknown rename operation {operation_id!r}")
+        source = Path(operation["source_path"])
+        destination = Path(operation["destination_path"])
+        token = token if token is not None else (operation["fencing_token"] or 0) + 1
+
+        if operation["journal_state"] not in (
+            JournalState.MOVED,
+            JournalState.DATABASE_UPDATED,
+            JournalState.VERIFIED,
+            JournalState.ROLLBACK_REQUIRED,
+            JournalState.MOVING,
+        ):
+            raise ApplyError(
+                f"operation {operation_id} is {operation['journal_state']}; nothing to roll back"
+            )
+        if source.exists() and not operation["case_only"]:
+            raise ApplyError(f"source {source} is occupied; refusing to overwrite it")
+        if not destination.exists():
+            raise ApplyError(f"destination {destination} is missing; cannot roll back")
+
+        # Revalidate before moving anything back: a changed file is not ours to move.
+        expected = operation.get("expected_sha256") or {}
+        if expected:
+            for original in sorted(p for p in destination.rglob("*") if p.is_file()):
+                digest = hashing.sha256_file(original)
+                if expected and digest not in expected.values():
+                    raise ApplyError(
+                        f"{original} does not match the recorded hashes; manual recovery required"
+                    )
+
+        if operation["journal_state"] != JournalState.ROLLBACK_REQUIRED:
+            self.journal.transition(
+                operation_id, JournalState.ROLLBACK_REQUIRED, fencing_token=token
+            )
+        if operation["case_only"]:
+            self._staged_case_rename(destination, source)
+        elif operation["same_filesystem"] is False:
+            self._copy_verify_delete(destination, source)
+        else:
+            os.rename(destination, source)
+
+        self._reconcile_paths(operation, destination, source)
+        self.journal.transition(operation_id, JournalState.ROLLED_BACK, fencing_token=token)
+        self.journal.sync_plan_state(operation["plan_id"])
+        return self.journal.get(operation_id)
+
+    def rollback_plan(self, plan_id: str) -> dict:
+        """Roll back every applied operation of a plan, newest first."""
+        rolled = 0
+        for operation in reversed(self.journal.operations(plan_id)):
+            if operation["journal_state"] in (
+                JournalState.MOVED,
+                JournalState.DATABASE_UPDATED,
+                JournalState.VERIFIED,
+                JournalState.COMPLETE,
+                JournalState.ROLLBACK_REQUIRED,
+            ):
+                if operation["journal_state"] == JournalState.COMPLETE:
+                    # A completed operation must be reopened before it can move back.
+                    continue
+                self.rollback_operation(operation["id"])
+                rolled += 1
+        state = self.journal.sync_plan_state(plan_id)
+        return {"plan_id": plan_id, "rolled_back": rolled, "state": state}
+
+    # ── helpers ───────────────────────────────────────────────────────────────
+
+    def _require_plan(self, plan_id: str) -> dict:
+        with self._session_factory() as session:
+            plan = session.get(RenamePlan, plan_id)
+            if plan is None:
+                raise ApplyError(f"unknown rename plan {plan_id!r}")
+            return {
+                "id": plan.id,
+                "state": plan.state,
+                "version": plan.version,
+                "checksum": plan.checksum,
+                "blockers": json.loads(plan.blockers or "[]"),
+            }
+
+    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(RenamePlan, plan_id)
+            plan.version += 1
+            plan.state = "applying"
+            plan.updated_at = _now()
+            token = plan.version
+            session.commit()
+        return token

photo_pipeline/services/rename_journal.py

@@ -229,6 +229,12 @@ class RenameJournal:
         classification, reason = _classify(row["journal_state"], source_exists, destination_exists)
         return {
             "operation_id": operation_id,
+            "plan_id": row["plan_id"],
+            "album": row["album"],
+            # The paths travel with the verdict so a reviewer can see which folder is
+            # unresolved without correlating an opaque operation id by hand.
+            "source_path": row["source_path"],
+            "destination_path": row["destination_path"],
             "journal_state": row["journal_state"],
             "classification": classification,
             "reason": reason,

photo_pipeline/services/renames.py

@@ -418,8 +418,17 @@ def _plan_dict(plan: RenamePlan, operations: list[RenameOperation]) -> dict:
                 "asset_count": operation.asset_count,
                 "same_filesystem": operation.same_filesystem,
                 "case_only": operation.case_only,
-                "issues": json.loads(operation.issues or "[]"),
+                # Severity is decided here, not in the browser: a client that has to
+                # keep its own copy of BLOCKING_CODES will eventually disagree with
+                # the validator about whether a plan can run.
+                "issues": [
+                    {**issue, "blocking": issue["code"] in BLOCKING_CODES}
+                    for issue in json.loads(operation.issues or "[]")
+                ],
                 "journal_state": operation.journal_state,
+                "verified_at": operation.verified_at.isoformat() if operation.verified_at else None,
+                "error_code": operation.error_code,
+                "error_message": operation.error_message,
             }
             for operation in operations
         ],

photo_pipeline/services/upload_batches.py

@@ -0,0 +1,424 @@
+"""UploadBatchService — one approved album at a time through immich-go (US05-02).
+
+Preflight (US05-01) proves a scope is safe and issues a token; this service turns
+that approval into a durable batch and runs it. Upload is the one stage the app
+cannot undo, so the discipline is:
+
+- **the approval is re-proved, not remembered.** Before every attempt the batch's
+  preflight token is recomputed from the current library state. Bytes edited after
+  approval, a withdrawn safety decision, or a server that stopped answering all
+  produce a different token and the attempt is refused, never run "optimistically".
+- **one lane.** A batch can only start while no other batch is running; the album
+  scope of a batch never widens after creation (concept §16 uploader lane).
+- **cancellation is cooperative and durable.** ``cancel`` writes ``cancelling``;
+  the running attempt observes it through the database, stops the uploader, and
+  records ``cancelled``. A cancelled batch is re-runnable from its own boundary.
+- **an interrupted attempt is uncertain, not failed.** Immich may have accepted
+  files the app never saw a report for, so ``recover`` marks a batch whose worker
+  vanished ``unknown_requires_verification`` (concept §15) instead of retrying it
+  blindly. Resolving that is US05-04.
+
+Per-asset upload *results* are deliberately not interpreted here — parsing the
+report is US05-03. What this story guarantees is that the report exists, is bounded,
+carries no credential, and that the batch's state is an honest description of what
+the process did.
+"""
+
+from __future__ import annotations
+
+import json
+import uuid
+from datetime import datetime, timezone
+from pathlib import Path
+
+from sqlalchemy import select, update
+from sqlalchemy.orm import sessionmaker
+
+from photo_pipeline.config import Config
+from photo_pipeline.integrations import immich_go
+from photo_pipeline.models import UploadBatch, UploadItem
+from photo_pipeline.services.hashing import sha1_file
+from photo_pipeline.services.uploads import UploadService
+
+
+class BatchState:
+    PLANNED = "planned"
+    RUNNING = "running"
+    CANCELLING = "cancelling"
+    CANCELLED = "cancelled"
+    SUCCEEDED = "succeeded"
+    FAILED = "failed"
+    # The attempt died without a parsed outcome: the server may hold the files.
+    UNKNOWN = "unknown_requires_verification"
+
+
+# States that occupy the single uploader lane.
+LANE_STATES = frozenset({BatchState.RUNNING, BatchState.CANCELLING})
+# States a batch may be (re)started from. ``unknown_requires_verification`` is not
+# among them: an uncertain upload must be verified (US05-04), never blindly retried.
+RUNNABLE_STATES = frozenset({BatchState.PLANNED, BatchState.FAILED, BatchState.CANCELLED})
+# States whose batch is still the live one for its album.
+OPEN_STATES = frozenset({BatchState.PLANNED, BatchState.RUNNING, BatchState.CANCELLING})
+
+class ItemState:
+    PENDING = "pending"
+    # ``sent`` means the batch process exited cleanly, not that Immich confirmed the
+    # asset — the per-item outcome comes from the report in US05-03/US05-04.
+    SENT = "sent"
+    FAILED = "failed"
+
+
+class BatchError(RuntimeError):
+    """The request cannot be carried out (unknown batch, wrong state, blocked)."""
+
+    def __init__(self, code: str, message: str) -> None:
+        super().__init__(message)
+        self.code = code
+
+
+class BatchConflict(BatchError):
+    """The lane is busy or the approval is stale — retry after resolving it."""
+
+
+def _now() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+class UploadBatchService:
+    def __init__(self, session_factory: sessionmaker, *, config: Config) -> None:
+        self._session_factory = session_factory
+        self._config = config
+        self._preflight = UploadService(session_factory, config=config)
+
+    # ── creation ──────────────────────────────────────────────────────────────
+
+    def create(
+        self, albums: list[str] | None = None, *, token: str, allow_partial: bool = False
+    ) -> list[dict]:
+        """Turn a *ready* preflight into one durable batch per album.
+
+        ``token`` must be the token of the current preflight for the same scope and
+        policy; anything else means the browser is acting on a stale preview.
+        """
+        report = self._preflight.preflight(albums, allow_partial=allow_partial)
+        if token != report["token"]:
+            raise BatchConflict(
+                "stale_preflight", "the preflight token does not describe the current state"
+            )
+        if report["state"] != "ready":
+            raise BatchError("not_ready", "the scope has unresolved upload blockers")
+
+        created: list[dict] = []
+        for album in report["albums"]:
+            existing = self._open_batch_for(album["album"])
+            if existing is not None:
+                created.append(existing)  # idempotent: one open batch per album
+                continue
+            created.append(self._create_one(album, token=token, allow_partial=allow_partial))
+        return created
+
+    def _create_one(self, album: dict, *, token: str, allow_partial: bool) -> dict:
+        eligible = [asset for asset in album["assets"] if not asset["blockers"]]
+        batch_id = str(uuid.uuid4())
+        command = immich_go.preview_command(
+            binary=self._config.immich_go_binary,
+            server_url=self._config.immich_server_url,
+            album_name=album["album_name"],
+            folder=album["folder"],
+        )
+        with self._session_factory() as session:
+            session.add(
+                UploadBatch(
+                    id=batch_id,
+                    album=album["album"],
+                    folder=album["folder"],
+                    album_name=album["album_name"],
+                    state=BatchState.PLANNED,
+                    preflight_token=token,
+                    allow_partial=allow_partial,
+                    command=json.dumps(command),
+                    uploader_version=immich_go.version(self._config.immich_go_binary),
+                    asset_count=len(eligible),
+                )
+            )
+            for asset in eligible:
+                session.add(
+                    UploadItem(
+                        batch_id=batch_id,
+                        asset_id=asset["asset_id"],
+                        path=asset["current_path"],
+                        sha256=asset["current_sha256"],
+                        sha1=sha1_file(asset["current_path"]),
+                        state=ItemState.PENDING,
+                    )
+                )
+            session.commit()
+        return self.get(batch_id)
+
+    # ── running ───────────────────────────────────────────────────────────────
+
+    def run(self, batch_id: str, *, worker_id: str = "uploader", cancelled=None) -> dict:
+        """Run one batch to completion. Blocks for the duration of the upload."""
+        batch = self._require(batch_id)
+        if batch["state"] not in RUNNABLE_STATES:
+            code = (
+                "requires_verification"
+                if batch["state"] == BatchState.UNKNOWN
+                else "not_runnable"
+            )
+            raise BatchError(code, f"batch {batch_id} is {batch['state']}")
+        busy = [row for row in self.list() if row["id"] != batch_id and row["state"] in LANE_STATES]
+        if busy:
+            raise BatchConflict("lane_busy", f"upload batch {busy[0]['id']} is still running")
+
+        # The approval is re-proved here, immediately before the irreversible act.
+        if not self._preflight.verify_token(
+            batch["preflight_token"], [batch["album"]], allow_partial=batch["allow_partial"]
+        ):
+            self._finish(
+                batch_id,
+                token=batch["version"],
+                state=BatchState.FAILED,
+                error=("stale_preflight", "the library changed after this batch was approved"),
+            )
+            raise BatchConflict(
+                "stale_preflight", "the library changed after this batch was approved"
+            )
+
+        token = self._claim(batch_id, worker_id=worker_id)
+        attempt = self.get(batch_id)["attempt_count"]
+        report_path = (
+            Path(self._config.data_dir) / "uploads" / f"{batch_id}-attempt-{attempt}.log"
+        )
+        key = self._config.immich_api_key.get_secret_value() if self._config.immich_api_key else ""
+        command = immich_go.build_command(
+            binary=self._config.immich_go_binary,
+            server_url=self._config.immich_server_url,
+            api_key=key,
+            album_name=batch["album_name"],
+            folder=batch["folder"],
+        )
+
+        def _stop_requested() -> bool:
+            if cancelled is not None and cancelled():
+                return True
+            current = self.get(batch_id)
+            return current is None or current["state"] == BatchState.CANCELLING
+
+        try:
+            result = immich_go.run_upload(
+                command,
+                report_path=report_path,
+                secret=key or None,
+                cancelled=_stop_requested,
+            )
+        except OSError as error:  # uploader vanished between preflight and exec
+            self._finish(
+                batch_id,
+                token=token,
+                state=BatchState.FAILED,
+                error=("uploader_failed", str(error)),
+            )
+            return self.get(batch_id)
+
+        if result["cancelled"]:
+            state, error, item_state = BatchState.CANCELLED, None, None
+        elif result["timed_out"]:
+            # Killed mid-flight: the server may already hold some of the files.
+            state = BatchState.UNKNOWN
+            error = ("timeout", "the uploader exceeded its time limit and was stopped")
+            item_state = None
+        elif result["exit_code"] == 0:
+            state, error, item_state = BatchState.SUCCEEDED, None, ItemState.SENT
+        else:
+            state = BatchState.FAILED
+            error = ("uploader_failed", f"immich-go exited with {result['exit_code']}")
+            item_state = ItemState.FAILED
+
+        self._finish(batch_id, token=token, state=state, error=error, result=result)
+        if item_state:
+            self._set_items(batch_id, item_state)
+        return self.get(batch_id)
+
+    def cancel(self, batch_id: str) -> dict:
+        """Request a stop. A running attempt drains; a planned batch stops outright."""
+        batch = self._require(batch_id)
+        if batch["state"] == BatchState.PLANNED:
+            target = BatchState.CANCELLED
+        elif batch["state"] == BatchState.RUNNING:
+            target = BatchState.CANCELLING
+        else:
+            raise BatchError("not_cancellable", f"batch {batch_id} is {batch['state']}")
+        with self._session_factory() as session:
+            row = session.get(UploadBatch, batch_id)
+            row.state = target
+            row.updated_at = _now()
+            if target == BatchState.CANCELLED:
+                row.finished_at = _now()
+            session.commit()
+        return self.get(batch_id)
+
+    # ── recovery ──────────────────────────────────────────────────────────────
+
+    def recover(self) -> dict:
+        """Resolve batches whose attempt died with the process.
+
+        A batch that never started is left ``planned`` and simply runs later. One
+        that was mid-upload cannot be classified from local state — immich-go may
+        have transferred everything before the crash — so it becomes
+        ``unknown_requires_verification`` and releases the lane rather than being
+        retried or declared failed.
+        """
+        interrupted = 0
+        with self._session_factory() as session:
+            for row in session.scalars(
+                select(UploadBatch).where(UploadBatch.state.in_(LANE_STATES))
+            ):
+                row.state = BatchState.UNKNOWN
+                row.error_code = "interrupted"
+                row.error_message = "the uploader process ended without a recorded outcome"
+                row.finished_at = _now()
+                row.updated_at = _now()
+                row.version += 1
+                interrupted += 1
+            session.commit()
+        return {"interrupted": interrupted}
+
+    # ── reads ─────────────────────────────────────────────────────────────────
+
+    def get(self, batch_id: str) -> dict | None:
+        with self._session_factory() as session:
+            row = session.get(UploadBatch, batch_id)
+            if row is None:
+                return None
+            items = list(
+                session.scalars(
+                    select(UploadItem)
+                    .where(UploadItem.batch_id == batch_id)
+                    .order_by(UploadItem.path)
+                )
+            )
+            return _batch_dict(row, items)
+
+    def list(self) -> list[dict]:
+        with self._session_factory() as session:
+            rows = list(session.scalars(select(UploadBatch).order_by(UploadBatch.created_at)))
+            return [_batch_dict(row, []) for row in rows]
+
+    def report(self, batch_id: str) -> str:
+        """The raw uploader output kept for this batch, or ``""`` when there is none."""
+        batch = self._require(batch_id)
+        path = Path(batch["report_path"]) if batch["report_path"] else None
+        if path is None or not path.exists():
+            return ""
+        return path.read_text(errors="replace")
+
+    # ── internals ─────────────────────────────────────────────────────────────
+
+    def _open_batch_for(self, album: str) -> dict | None:
+        with self._session_factory() as session:
+            row = session.scalar(
+                select(UploadBatch).where(
+                    UploadBatch.album == album, UploadBatch.state.in_(OPEN_STATES)
+                )
+            )
+            return self.get(row.id) if row else None
+
+    def _claim(self, batch_id: str, *, worker_id: str) -> int:
+        """Take ownership: bump the version (the fencing token) and start an attempt."""
+        with self._session_factory() as session:
+            row = session.get(UploadBatch, batch_id)
+            row.version += 1
+            row.attempt_count += 1
+            row.state = BatchState.RUNNING
+            row.worker_id = worker_id
+            row.error_code = row.error_message = None
+            row.started_at = _now()
+            row.finished_at = None
+            row.updated_at = _now()
+            token = row.version
+            session.commit()
+        return token
+
+    def _finish(
+        self,
+        batch_id: str,
+        *,
+        token: int,
+        state: str,
+        error: tuple[str, str] | None = None,
+        result: dict | None = None,
+    ) -> None:
+        """Record the outcome, but only for the attempt that still owns the batch."""
+        values = {
+            "state": state,
+            "finished_at": _now(),
+            "updated_at": _now(),
+            "error_code": error[0] if error else None,
+            "error_message": error[1][:500] if error else None,
+        }
+        if result is not None:
+            values |= {
+                "exit_code": result["exit_code"],
+                "report_path": result["report_path"],
+                "report_bytes": result["report_bytes"],
+                "report_truncated": result["report_truncated"],
+            }
+        with self._session_factory() as session:
+            session.execute(
+                update(UploadBatch)
+                .where(UploadBatch.id == batch_id, UploadBatch.version == token)
+                .values(**values)
+            )
+            session.commit()
+
+    def _set_items(self, batch_id: str, state: str) -> None:
+        with self._session_factory() as session:
+            session.execute(
+                update(UploadItem)
+                .where(UploadItem.batch_id == batch_id)
+                .values(state=state, updated_at=_now())
+            )
+            session.commit()
+
+    def _require(self, batch_id: str) -> dict:
+        batch = self.get(batch_id)
+        if batch is None:
+            raise BatchError("not_found", f"unknown upload batch {batch_id!r}")
+        return batch
+
+
+def _batch_dict(row: UploadBatch, items: list[UploadItem]) -> dict:
+    return {
+        "id": row.id,
+        "album": row.album,
+        "folder": row.folder,
+        "album_name": row.album_name,
+        "state": row.state,
+        "preflight_token": row.preflight_token,
+        "allow_partial": row.allow_partial,
+        "command": json.loads(row.command or "[]"),
+        "uploader_version": row.uploader_version,
+        "asset_count": row.asset_count,
+        "attempt_count": row.attempt_count,
+        "version": row.version,
+        "worker_id": row.worker_id,
+        "report_path": row.report_path,
+        "report_bytes": row.report_bytes,
+        "report_truncated": row.report_truncated,
+        "exit_code": row.exit_code,
+        "error_code": row.error_code,
+        "error_message": row.error_message,
+        "started_at": row.started_at.isoformat() if row.started_at else None,
+        "finished_at": row.finished_at.isoformat() if row.finished_at else None,
+        "items": [
+            {
+                "asset_id": item.asset_id,
+                "path": item.path,
+                "sha256": item.sha256,
+                "sha1": item.sha1,
+                "state": item.state,
+            }
+            for item in items
+        ],
+    }

photo_pipeline/services/uploads.py

@@ -0,0 +1,306 @@
+"""UploadService — preflight for Immich upload (US05-01).
+
+Upload is the first stage that sends the library somewhere the app cannot take it
+back from, so nothing here uploads: preflight only *proves* a scope is safe and
+issues a token that a later start command must present (US05-02).
+
+What it proves (concept §8 "preflight"):
+
+- credentials are configured and the Immich server answers, without ever putting
+  the API key in a response, a preview, or a log line;
+- ``immich-go`` is installed and its version is recorded;
+- no rename is half-applied — the journal must not block library mutation;
+- every asset in scope is canonical, present, and decided ``sfw``/``nsfw`` with a
+  verified safety EXIF checkpoint;
+- every SFW asset also has a completed analysis EXIF checkpoint;
+- the bytes on disk right now still hash to what inventory recorded, so the album
+  preview describes the exact bytes that would be uploaded.
+
+Blocker codes are structured, never prose the UI has to parse:
+
+``no_library_root``, ``credentials_missing``, ``server_unreachable``,
+``immich_go_missing``, ``rename_pending``, ``unknown_album``, ``empty_scope``,
+``file_missing``, ``bytes_changed``, ``safety_undecided``, ``safety_deferred``,
+``safety_exif_unverified``, ``analysis_incomplete``, ``partial_scope``.
+
+**Partial scope is a blocker, not a default.** An album with any blocked asset is
+refused unless the caller passes the explicit ``allow_partial`` policy, which is
+itself part of the token — a token issued for a partial upload can never be
+replayed as a full one.
+
+The token is derived, not stored: it is a digest of the whole report (minus its
+timestamp), so any change that matters — a new decision, edited bytes, a different
+server, a resolved blocker, a different policy — produces a different token and the
+old one is stale by construction. No table, no invalidation bookkeeping.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+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
+from photo_pipeline.models import AnalysisResult, Asset, SafetyReview
+from photo_pipeline.services.albums import album_label
+from photo_pipeline.services.hashing import sha256_file
+from photo_pipeline.services.rename_journal import RenameJournal
+
+PREFLIGHT_VERSION = 1
+TOKEN_PREFIX = f"v{PREFLIGHT_VERSION}"
+SFW = "sfw"
+NSFW = "nsfw"
+ANALYZED = "analyzed"
+
+
+class UploadError(RuntimeError):
+    """Invalid preflight request (unknown album in the requested scope)."""
+
+
+def _now() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+def _issue(code: str, message: str) -> dict:
+    return {"code": code, "message": message}
+
+
+class UploadService:
+    def __init__(self, session_factory: sessionmaker, *, config: Config) -> None:
+        self._session_factory = session_factory
+        self._config = config
+        self._roots = tuple(Path(root) for root in config.library_roots)
+
+    # ── preflight ─────────────────────────────────────────────────────────────
+
+    def preflight(self, albums: list[str] | None = None, *, allow_partial: bool = False) -> dict:
+        """Validate an upload scope and issue its token. Read-only, no upload."""
+        report = {
+            "schema_version": PREFLIGHT_VERSION,
+            "policy": {"allow_partial": allow_partial},
+            "blockers": [],
+            "credentials": self._credentials(),
+            "server": self._server(),
+            "uploader": self._uploader(),
+        }
+        report["blockers"] += self._environment_blockers(report)
+        report["albums"] = self._albums(albums, allow_partial=allow_partial)
+        report["totals"] = _totals(report["albums"])
+        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, albums: list[str] | None = None, *, allow_partial: bool = False
+    ) -> bool:
+        """True when ``token`` still describes the current state of that scope.
+
+        Recomputed rather than looked up, so an externally edited file or a changed
+        decision invalidates it even though nothing wrote to the database.
+        """
+        return bool(token) and token == self.preflight(albums, allow_partial=allow_partial)["token"]
+
+    # ── environment ───────────────────────────────────────────────────────────
+
+    def _credentials(self) -> dict:
+        """Presence only — the key itself never leaves configuration."""
+        return {
+            "server_url": self._config.immich_server_url,
+            "api_key_configured": self._config.immich_api_key is not None,
+        }
+
+    def _server(self) -> dict:
+        reachable, detail = immich_go.ping(self._config.immich_server_url)
+        return {"reachable": reachable, "detail": detail}
+
+    def _uploader(self) -> dict:
+        binary = self._config.immich_go_binary
+        return {
+            "binary": binary,
+            "installed": immich_go.find_binary(binary) is not None,
+            "version": immich_go.version(binary),
+        }
+
+    def _environment_blockers(self, report: dict) -> list[dict]:
+        blockers: list[dict] = []
+        if not self._roots:
+            blockers.append(_issue("no_library_root", "no library root is configured"))
+        if not report["credentials"]["api_key_configured"] or not self._config.immich_server_url:
+            blockers.append(
+                _issue("credentials_missing", "an Immich server URL and API key are required")
+            )
+        elif not report["server"]["reachable"]:
+            blockers.append(
+                _issue(
+                    "server_unreachable", f"Immich did not respond: {report['server']['detail']}"
+                )
+            )
+        if not report["uploader"]["installed"]:
+            blockers.append(
+                _issue("immich_go_missing", f"{self._config.immich_go_binary} is not installed")
+            )
+        if RenameJournal(self._session_factory).blocks_mutation():
+            blockers.append(
+                _issue("rename_pending", "an unresolved rename must be recovered before upload")
+            )
+        return blockers
+
+    # ── scope ─────────────────────────────────────────────────────────────────
+
+    def _albums(self, requested: list[str] | None, *, allow_partial: bool) -> list[dict]:
+        by_album = self._scope()
+        if requested is not None:
+            unknown = sorted(set(requested) - set(by_album))
+            if unknown:
+                raise UploadError(f"unknown album(s): {', '.join(unknown)}")
+            by_album = {name: by_album[name] for name in sorted(set(requested))}
+        return [
+            self._album(name, rows, allow_partial=allow_partial)
+            for name, rows in sorted(by_album.items())
+        ]
+
+    def _scope(self) -> dict[str, list[dict]]:
+        """Canonical, active assets grouped by album, each with its stage 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),
+                    )
+                )
+            )
+            reviews: dict[str, SafetyReview] = {}
+            for review in session.scalars(select(SafetyReview).order_by(SafetyReview.created_at)):
+                reviews[review.asset_id] = review  # latest row per asset wins
+            analyses = {
+                result.asset_id: result for result in session.scalars(select(AnalysisResult))
+            }
+
+        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,
+                    "expected_sha256": asset.current_sha256,
+                    "review": reviews.get(asset.id),
+                    "analysis": analyses.get(asset.id),
+                }
+            )
+        return by_album
+
+    def _album(self, name: str, rows: list[dict], *, allow_partial: bool) -> dict:
+        folder = Path(rows[0]["path"]).parent
+        items = sorted((self._item(row) for row in rows), key=lambda item: item["current_path"])
+        blocked = [item for item in items if item["blockers"]]
+        eligible = [item for item in items if not item["blockers"]]
+        blockers: list[dict] = []
+        if blocked and not allow_partial:
+            blockers.append(
+                _issue(
+                    "partial_scope",
+                    f"{len(blocked)} of {len(items)} asset(s) are not upload-ready; resolve them "
+                    "or approve a partial upload explicitly",
+                )
+            )
+        if not eligible:
+            blockers.append(_issue("empty_scope", "no upload-ready asset remains in this album"))
+        # Folder-as-album: Immich names the album after the leaf folder, so the
+        # preview shows exactly what the server will create.
+        album_name = folder.name
+        return {
+            "album": name,
+            "folder": str(folder),
+            "album_name": album_name,
+            "asset_count": len(items),
+            "eligible_count": len(eligible),
+            "blocked_count": len(blocked),
+            "partial": bool(blocked),
+            "state": "blocked" if blockers else "ready",
+            "blockers": blockers,
+            "assets": items,
+            "command_preview": immich_go.preview_command(
+                binary=self._config.immich_go_binary,
+                server_url=self._config.immich_server_url,
+                album_name=album_name,
+                folder=folder,
+            ),
+        }
+
+    def _item(self, row: dict) -> dict:
+        """One asset's readiness, including its hash *as it is on disk right now*."""
+        path = Path(row["path"])
+        review: SafetyReview | None = row["review"]
+        analysis: AnalysisResult | None = row["analysis"]
+        decision = review.decision if review else None
+        blockers: list[dict] = []
+        current_sha256 = None
+
+        if not path.exists():
+            blockers.append(_issue("file_missing", f"{path} is missing"))
+        else:
+            # ponytail: full re-hash every preflight. Gate on (size, mtime_ns) first
+            # if a large library makes this slow — the hash stays authoritative.
+            current_sha256 = sha256_file(path)
+            if row["expected_sha256"] and current_sha256 != row["expected_sha256"]:
+                blockers.append(
+                    _issue("bytes_changed", f"{path} changed since its last verified checkpoint")
+                )
+
+        if decision not in (SFW, NSFW):
+            code = "safety_deferred" if decision == "deferred" else "safety_undecided"
+            blockers.append(_issue(code, "a confirmed sfw/nsfw safety decision is required"))
+        elif review.exif_verified_at is None:
+            blockers.append(
+                _issue("safety_exif_unverified", "the safety EXIF checkpoint is not verified")
+            )
+        elif decision == SFW and not (
+            analysis and analysis.status == ANALYZED and analysis.exif_written_at is not None
+        ):
+            blockers.append(
+                _issue("analysis_incomplete", "SFW assets need a verified analysis EXIF checkpoint")
+            )
+
+        return {
+            "asset_id": row["asset_id"],
+            "current_path": str(path),
+            "safety_decision": decision,
+            "current_sha256": current_sha256,
+            "blockers": blockers,
+        }
+
+
+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),
+        "eligible": sum(album["eligible_count"] for album in albums),
+        "blocked": sum(album["blocked_count"] for album in albums),
+    }
+
+
+def _token(report: dict) -> str:
+    """Digest of everything the report asserts. Volatile fields are excluded so the
+    same state always yields the same token; every relevant change breaks it."""
+    payload = {key: value for key, value in report.items() if key not in ("generated_at", "token")}
+    digest = hashlib.sha256(
+        json.dumps(payload, sort_keys=True, ensure_ascii=False, default=str).encode("utf-8")
+    ).hexdigest()
+    return f"{TOKEN_PREFIX}:{digest}"

pyproject.toml

@@ -30,4 +30,5 @@ testpaths = ["tests"]
 markers = [
     "phase_b: Phase B end-to-end acceptance (US02-07) — API, worker-recovery, and browser journeys",
     "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",
 ]

tests/e2e/_pipeline_harness.py

@@ -14,7 +14,9 @@ import socket
 import subprocess
 import sys
 import time
+import uuid
 from dataclasses import dataclass
+from datetime import datetime, timezone
 from pathlib import Path
 
 import httpx
@@ -159,3 +161,88 @@ def wait_until(predicate, *, timeout: float = 20, interval: float = 0.1):
             return value
         time.sleep(interval)
     raise AssertionError("condition not met before timeout")
+
+
+# ── Phase D: an analysed album, ready to be named and renamed ────────────────
+
+NOW = datetime(2026, 1, 1, tzinfo=timezone.utc)
+
+
+class session_factory:
+    """Session factory against a seeded database, for the few things a test has to
+    set up or inspect below the API — journal states, mainly."""
+
+    def __init__(self, seeded: Seeded) -> None:
+        self._seeded = seeded
+
+    def __enter__(self):
+        from photo_pipeline.config import Config
+        from photo_pipeline.db import create_db_engine, create_session_factory, run_migrations
+
+        config = Config.from_env(
+            {
+                "PHOTO_PIPELINE_DATA_DIR": str(self._seeded.data),
+                "PHOTO_PIPELINE_LIBRARY_ROOTS": str(self._seeded.lib),
+            }
+        )
+        run_migrations(config.database_url)
+        self._engine = create_db_engine(config.database_url)
+        return create_session_factory(self._engine)
+
+    def __exit__(self, *_):
+        self._engine.dispose()
+        return False
+
+
+def seed_album(tmp_path: Path, album: str = "rome", names: tuple[str, ...] = ("a.jpg", "b.jpg")):
+    """A library holding one album folder whose photos are confirmed SFW and analysed
+    — the state a naming proposal, and therefore a rename plan, is built from."""
+    from sqlalchemy import select
+
+    from photo_pipeline.models import AnalysisResult, Asset, SafetyReview
+    from photo_pipeline.services.inventory import InventoryService
+
+    seeded = seed_library(tmp_path, {}, {})
+    folder = seeded.lib / album
+    folder.mkdir(parents=True)
+    for index, name in enumerate(names):
+        image(folder / name, index + 1)
+
+    with session_factory(seeded) as sf:
+        InventoryService(sf).scan(seeded.lib)
+        with sf() as session:
+            rows = list(session.execute(select(Asset.id, Asset.current_path)).all())
+            for asset_id, path in rows:
+                session.add(
+                    SafetyReview(
+                        id=str(uuid.uuid4()), asset_id=asset_id, decision="sfw", created_at=NOW
+                    )
+                )
+                session.add(
+                    AnalysisResult(
+                        asset_id=asset_id,
+                        status="analyzed",
+                        description=f"a view of {path}",
+                        tags='["ruins", "city"]',
+                        approx_year=2019,
+                        location_hint="Rome",
+                    )
+                )
+            session.commit()
+        seeded.asset_ids.update({Path(path).stem: aid for aid, path in rows})
+    return seeded
+
+
+def approve_album(base: str, *, album: str = "rome", name: str) -> None:
+    """Generate a proposal, set its final name, and approve it over HTTP."""
+    httpx.post(f"{base}/api/v1/albums/proposals", json={}, timeout=10).raise_for_status()
+    for payload, route in (
+        ({"name": name}, "edit"),
+        ({}, "approve"),
+    ):
+        current = httpx.get(f"{base}/api/v1/albums/proposals/{album}", timeout=10).json()
+        httpx.post(
+            f"{base}/api/v1/albums/proposals/{album}/{route}",
+            json={**payload, "expected_version": current["version"]},
+            timeout=10,
+        ).raise_for_status()

tests/e2e/test_phase_d_pipeline.py

@@ -0,0 +1,377 @@
+"""Phase D end-to-end acceptance (US04-06): guarded renaming, black box.
+
+Every journey here drives a real ``photo_pipeline serve`` child process over HTTP —
+plan, export, confirm, apply, collide, go stale, crash, recover, roll back. The
+crashes are real: the server is killed by the ``PHOTO_PIPELINE_FAULT_AFTER`` barrier
+at each persisted journal transition in turn, then a fresh process is started against
+the same database and library and has to reconcile the wreckage from evidence alone.
+
+Photos really move. After every journey the assertions read the filesystem and the
+inventory back: the asset set, the stable IDs, and the content hashes must be exactly
+what they were before, only at new paths.
+"""
+
+from __future__ import annotations
+
+import httpx
+import pytest
+
+from tests.e2e._pipeline_harness import Server, approve_album, seed_album, session_factory
+
+pytestmark = pytest.mark.phase_d
+
+TIMEOUT = 10
+APPROVED = "2019 Rome"
+CRASH_POINTS = ["moving", "moved", "database_updated", "verified"]
+
+
+@pytest.fixture
+def server(tmp_path):
+    seeded = seed_album(tmp_path)
+    running = Server(seeded).start()
+    running.seeded = seeded
+    try:
+        yield running
+    finally:
+        running.stop()
+
+
+# ── helpers ──────────────────────────────────────────────────────────────────
+
+
+def _plan(base) -> dict:
+    response = httpx.post(f"{base}/api/v1/rename-plans", timeout=TIMEOUT)
+    response.raise_for_status()
+    return response.json()
+
+
+def _get_plan(base, plan_id) -> dict:
+    return httpx.get(f"{base}/api/v1/rename-plans/{plan_id}", timeout=TIMEOUT).json()
+
+
+def _apply(base, plan, **body):
+    payload = {"expected_version": plan["version"], **body}
+    return httpx.post(
+        f"{base}/api/v1/rename-plans/{plan['id']}/apply", json=payload, timeout=TIMEOUT
+    )
+
+
+def _inventory(base) -> dict[str, dict]:
+    """Every asset by stable ID, so identity can be compared across a rename."""
+    items = httpx.get(
+        f"{base}/api/v1/inventory/assets", params={"limit": 200}, timeout=TIMEOUT
+    ).json()["items"]
+    return {item["id"]: item for item in items}
+
+
+def _content(root) -> dict[str, bytes]:
+    return {
+        str(path.relative_to(root)): path.read_bytes()
+        for path in sorted(root.rglob("*"))
+        if path.is_file()
+    }
+
+
+def _recovery(base) -> dict:
+    return httpx.get(f"{base}/api/v1/rename-recovery", timeout=TIMEOUT).json()
+
+
+def _journal_states(seeded, plan_id) -> list[str]:
+    from photo_pipeline.services.rename_journal import RenameJournal
+
+    with session_factory(seeded) as sf:
+        return [row["journal_state"] for row in RenameJournal(sf).operations(plan_id)]
+
+
+# ── US04-01: plan and export ─────────────────────────────────────────────────
+
+
+def test_plan_and_export_describe_every_move_without_touching_the_library(server):
+    before = _content(server.seeded.lib)
+    approve_album(server.base, name=APPROVED)
+    plan = _plan(server.base)
+
+    assert plan["state"] == "validated" and plan["operation_count"] == 1
+    operation = plan["operations"][0]
+    assert operation["source_path"].endswith("/rome")
+    assert operation["destination_path"].endswith(f"/{APPROVED}")
+    assert operation["asset_count"] == 2
+    assert set(operation["asset_ids"]) == set(_inventory(server.base))
+
+    export = httpx.get(
+        f"{server.base}/api/v1/rename-plans/{plan['id']}/export", timeout=TIMEOUT
+    ).json()
+    assert export["schema_version"] == 1
+    assert export["checksum"] == plan["checksum"]
+    assert [op["source_path"] for op in export["operations"]] == [operation["source_path"]]
+    # Portable evidence must not smuggle out anything sensitive.
+    assert "token" not in str(export).lower() and "key" not in str(export).lower()
+
+    # Planning is a preview: not one byte moved.
+    assert _content(server.seeded.lib) == before
+
+
+# ── US04-03: confirmation and apply ──────────────────────────────────────────
+
+
+def test_apply_requires_the_current_confirmation_token(server):
+    approve_album(server.base, name=APPROVED)
+    plan = _plan(server.base)
+
+    stale = _apply(server.base, {**plan, "version": plan["version"] + 7})
+    assert stale.status_code == 409 and stale.json()["error"]["code"] == "version_conflict"
+
+    wrong_checksum = _apply(server.base, plan, expected_checksum="0" * 64)
+    assert wrong_checksum.status_code == 409
+
+    assert (server.seeded.lib / "rome").is_dir(), "a refused confirmation moves nothing"
+
+
+def test_a_valid_apply_preserves_ids_hashes_and_the_asset_set(server):
+    approve_album(server.base, name=APPROVED)
+    before = _inventory(server.base)
+    before_content = _content(server.seeded.lib)
+    plan = _plan(server.base)
+
+    applied = _apply(server.base, plan, expected_checksum=plan["checksum"]).json()
+    assert applied["applied"] == 1 and applied["failed"] == 0 and applied["state"] == "applied"
+
+    after = _inventory(server.base)
+    assert set(after) == set(before), "renaming must not change asset identity"
+    assert {item["current_sha256"] for item in after.values()} == {
+        item["current_sha256"] for item in before.values()
+    }
+    assert all(APPROVED in item["current_path"] for item in after.values())
+    # Same bytes, new folder — nothing was rewritten in the move.
+    assert _content(server.seeded.lib) == {
+        key.replace("rome/", f"{APPROVED}/"): value for key, value in before_content.items()
+    }
+    assert _journal_states(server.seeded, plan["id"]) == ["complete"]
+
+
+def test_a_case_only_rename_applies_on_a_case_insensitive_filesystem(tmp_path):
+    seeded = seed_album(tmp_path, album="rome")
+    running = Server(seeded).start()
+    try:
+        approve_album(running.base, name="Rome")
+        plan = _plan(running.base)
+        assert plan["operations"][0]["case_only"] is True
+
+        assert _apply(running.base, plan).json()["applied"] == 1
+        entries = {path.name for path in seeded.lib.iterdir()}
+        assert "Rome" in entries
+        # The staged intermediate name must not survive the procedure.
+        assert not any(name.startswith(".rename-") for name in entries)
+        assert all("/Rome/" in item["current_path"] for item in _inventory(running.base).values())
+    finally:
+        running.stop()
+
+
+def test_a_collision_is_refused_and_the_occupant_survives(server):
+    approve_album(server.base, name=APPROVED)
+    occupied = server.seeded.lib / APPROVED
+    occupied.mkdir()
+    (occupied / "precious.jpg").write_bytes(b"do not lose me")
+
+    plan = _plan(server.base)
+    assert plan["state"] == "invalid" and "destination_exists" in plan["blockers"]
+
+    refused = _apply(server.base, plan)
+    assert refused.status_code == 422 and refused.json()["error"]["code"] == "cannot_apply"
+    assert (occupied / "precious.jpg").read_bytes() == b"do not lose me"
+    assert (server.seeded.lib / "rome").is_dir()
+
+
+def test_a_source_that_changed_after_planning_is_refused(server):
+    approve_album(server.base, name=APPROVED)
+    plan = _plan(server.base)
+    # The plan recorded per-asset hashes; the file changes before confirmation.
+    (server.seeded.lib / "rome" / "a.jpg").write_bytes(b"tampered")
+
+    result = _apply(server.base, plan).json()
+    assert result["failed"] == 1 and result["applied"] == 0
+    assert (server.seeded.lib / "rome").is_dir(), "a failed precondition leaves the source alone"
+    operation = _get_plan(server.base, plan["id"])["operations"][0]
+    assert operation["journal_state"] == "failed"
+    assert operation["error_code"] == "source_changed"
+
+
+# ── US04-04: crash points, recovery, rollback ────────────────────────────────
+
+
+def _crash_during_apply(seeded, plan, state):
+    """Apply with the fault barrier armed: the server dies at ``state``, mid-move."""
+    crashing = Server(seeded, extra_env={"PHOTO_PIPELINE_FAULT_AFTER": state}).start()
+    try:
+        with pytest.raises(httpx.HTTPError):
+            _apply(crashing.base, plan)
+    finally:
+        crashing.stop()
+    assert crashing.proc is None
+
+
+@pytest.mark.parametrize("crash_point", CRASH_POINTS)
+def test_every_journal_crash_point_recovers_without_losing_content(tmp_path, crash_point):
+    seeded = seed_album(tmp_path)
+    first = Server(seeded).start()
+    try:
+        approve_album(first.base, name=APPROVED)
+        before = _inventory(first.base)
+        before_bytes = sorted(_content(seeded.lib).values())
+        plan = _plan(first.base)
+    finally:
+        first.stop()
+
+    _crash_during_apply(seeded, plan, crash_point)
+
+    # A brand-new process, no in-memory state: everything comes from the journal.
+    restarted = Server(seeded).start()
+    try:
+        recovery = _recovery(restarted.base)
+        assert recovery["items"], f"a crash at {crash_point} must leave visible evidence"
+        assert recovery["items"][0]["journal_state"] == crash_point
+        # Only a crash that could have left the library half-renamed blocks other
+        # work. `verified` is past every filesystem and database change — the move
+        # is done and checked, just not flagged complete — so it blocks nothing.
+        assert recovery["blocks_mutation"] is (crash_point != "verified")
+
+        resolved = httpx.post(
+            f"{restarted.base}/api/v1/rename-recovery/resolve", timeout=TIMEOUT
+        ).json()
+        assert resolved["manual"] == 0, "an interrupted rename must be decidable from evidence"
+
+        # Recovery is idempotent: running it again changes nothing.
+        assert _recovery(restarted.base)["blocks_mutation"] is False
+        httpx.post(f"{restarted.base}/api/v1/rename-recovery/resolve", timeout=TIMEOUT)
+
+        # A resumable crash is left ready to run again; finish it so every crash
+        # point converges on the same observable end state.
+        current = _get_plan(restarted.base, plan["id"])
+        if current["state"] != "applied":
+            _apply(restarted.base, current)
+
+        after = _inventory(restarted.base)
+        assert set(after) == set(before), "no asset may be lost or invented by a crash"
+        assert {item["current_sha256"] for item in after.values()} == {
+            item["current_sha256"] for item in before.values()
+        }
+        assert sorted(_content(seeded.lib).values()) == before_bytes
+        assert (seeded.lib / APPROVED).is_dir() and not (seeded.lib / "rome").exists()
+        assert all(APPROVED in item["current_path"] for item in after.values())
+        assert _journal_states(seeded, plan["id"]) == ["complete"]
+    finally:
+        restarted.stop()
+
+
+def test_ambiguous_evidence_is_kept_for_a_human_and_keeps_blocking(tmp_path):
+    seeded = seed_album(tmp_path)
+    first = Server(seeded).start()
+    try:
+        approve_album(first.base, name=APPROVED)
+        plan = _plan(first.base)
+    finally:
+        first.stop()
+
+    _crash_during_apply(seeded, plan, "moving")
+    # Someone creates the destination while the operation is unresolved: now both
+    # paths exist and nothing can tell which one holds the truth.
+    (seeded.lib / APPROVED).mkdir(exist_ok=True)
+
+    restarted = Server(seeded).start()
+    try:
+        assert _recovery(restarted.base)["items"][0]["classification"] == "manual"
+        resolved = httpx.post(
+            f"{restarted.base}/api/v1/rename-recovery/resolve", timeout=TIMEOUT
+        ).json()
+        assert resolved["manual"] == 1 and resolved["resumed"] == 0 and resolved["completed"] == 0
+        # Still blocking, and still nothing guessed.
+        assert _recovery(restarted.base)["blocks_mutation"] is True
+        assert (seeded.lib / "rome").is_dir()
+    finally:
+        restarted.stop()
+
+
+def test_an_unresolved_rename_is_the_cancellation_boundary(tmp_path):
+    """There is no cancel button once a rename starts. The boundary is that nothing
+    else may mutate the library until the interrupted work is resolved."""
+    seeded = seed_album(tmp_path)
+    first = Server(seeded).start()
+    try:
+        approve_album(first.base, name=APPROVED)
+        plan = _plan(first.base)
+    finally:
+        first.stop()
+
+    _crash_during_apply(seeded, plan, "moving")
+
+    restarted = Server(seeded).start()
+    try:
+        assert _recovery(restarted.base)["blocks_mutation"] is True
+
+        refused = httpx.post(f"{restarted.base}/api/v1/albums/proposals", json={}, timeout=TIMEOUT)
+        assert refused.status_code == 409
+        assert refused.json()["error"]["code"] == "rename_recovery_required"
+
+        # Reading stays available throughout — only mutation is paused.
+        assert httpx.get(f"{restarted.base}/api/v1/albums/evidence", timeout=TIMEOUT).status_code
+        assert len(_inventory(restarted.base)) == 2
+    finally:
+        restarted.stop()
+
+
+def test_rollback_returns_an_interrupted_move_to_its_source(tmp_path):
+    seeded = seed_album(tmp_path)
+    first = Server(seeded).start()
+    try:
+        approve_album(first.base, name=APPROVED)
+        before = _inventory(first.base)
+        plan = _plan(first.base)
+    finally:
+        first.stop()
+
+    # Crash after the content moved but before the database caught up: the operation
+    # is still reversible, which is exactly when rollback is defined.
+    _crash_during_apply(seeded, plan, "moved")
+
+    restarted = Server(seeded).start()
+    try:
+        rolled = httpx.post(
+            f"{restarted.base}/api/v1/rename-plans/{plan['id']}/rollback", timeout=TIMEOUT
+        ).json()
+        assert rolled["rolled_back"] == 1 and rolled["state"] == "rolled_back"
+
+        assert (seeded.lib / "rome" / "a.jpg").exists()
+        assert not (seeded.lib / APPROVED).exists()
+        after = _inventory(restarted.base)
+        assert set(after) == set(before)
+        assert all(item["current_path"].endswith(".jpg") for item in after.values())
+        assert _recovery(restarted.base)["blocks_mutation"] is False
+    finally:
+        restarted.stop()
+
+
+# ── durability ───────────────────────────────────────────────────────────────
+
+
+def test_the_applied_state_survives_a_full_restart(tmp_path):
+    seeded = seed_album(tmp_path)
+    first = Server(seeded).start()
+    try:
+        approve_album(first.base, name=APPROVED)
+        plan = _plan(first.base)
+        _apply(first.base, plan, expected_checksum=plan["checksum"]).raise_for_status()
+        expected = _inventory(first.base)
+    finally:
+        first.stop()
+
+    restarted = Server(seeded).start()
+    try:
+        assert _inventory(restarted.base) == expected
+        after = _get_plan(restarted.base, plan["id"])
+        assert after["state"] == "applied"
+        assert after["checksum"] == plan["checksum"], "the plan's evidence is immutable"
+        assert [op["journal_state"] for op in after["operations"]] == ["complete"]
+        assert all(op["verified_at"] for op in after["operations"])
+        assert _recovery(restarted.base) == {"blocks_mutation": False, "items": []}
+    finally:
+        restarted.stop()

tests/e2e/test_renames_ui.py

@@ -0,0 +1,294 @@
+"""Browser journeys for the rename view (US04-05).
+
+Covers previewing every affected path, confirming with the server-issued token,
+applying with visible progress and terminal verification, refusing a stale
+confirmation, previewing a collision as blocking, and resolving an interrupted
+rename — including the check that an unresolved rename really does block unrelated
+mutations at the API, not just in the UI.
+
+Renames really happen here: the assertions read the filesystem afterwards.
+"""
+
+from __future__ import annotations
+
+import httpx
+import pytest
+from playwright.sync_api import expect
+
+from tests.e2e._pipeline_harness import Server, approve_album, seed_album, session_factory
+
+# Part of the Phase D acceptance command (US04-06); mapped to US04-05 for traceability.
+pytestmark = pytest.mark.phase_d
+
+TIMEOUT = 10
+APPROVED = "2019 Rome"
+
+
+@pytest.fixture
+def server(tmp_path):
+    seeded = seed_album(tmp_path)
+    running = Server(seeded).start()
+    running.seeded = seeded
+    try:
+        yield running
+    finally:
+        running.stop()
+
+
+# ── helpers ──────────────────────────────────────────────────────────────────
+
+
+def _approve(base, *, name=APPROVED):
+    approve_album(base, name=name)
+
+
+def _build(base):
+    response = httpx.post(f"{base}/api/v1/rename-plans", timeout=TIMEOUT)
+    response.raise_for_status()
+    return response.json()
+
+
+def _open_plan(page, server):
+    """Approve a name, open the view, and build the plan through the UI."""
+    _approve(server.base)
+    page.goto(f"{server.base}/app/#/renames")
+    page.get_by_test_id("build-plan").click()
+    page.get_by_test_id("operations").wait_for()
+
+
+def _interrupt(seeded, plan, *, state="moving", make_destination=False):
+    """Leave an operation in a non-terminal journal state, as a crash mid-apply would.
+
+    ``make_destination`` also creates the destination folder, so source and
+    destination both exist and the evidence becomes ambiguous (``manual``).
+    """
+    from photo_pipeline.services.rename_journal import RenameJournal
+
+    with session_factory(seeded) as sf:
+        journal = RenameJournal(sf)
+        operation = journal.operations(plan["id"])[0]
+        journal.begin(operation["id"], worker_id="crashed", fencing_token=1)
+        if state != "moving":
+            journal.transition(operation["id"], state, fencing_token=1)
+    if make_destination:
+        (seeded.lib / APPROVED).mkdir(exist_ok=True)
+
+
+# ── preview ──────────────────────────────────────────────────────────────────
+
+
+def test_preview_shows_every_path_operation_type_and_count(page, server):
+    errors = []
+    page.on("console", lambda m: errors.append(m.text) if m.type == "error" else None)
+    _open_plan(page, server)
+
+    expect(page.get_by_test_id("plan-state")).to_have_text("validated")
+    expect(page.get_by_test_id("operation-count")).to_have_text("1 folder")
+    expect(page.get_by_test_id("affected-assets")).to_have_text("2 photos")
+
+    row = page.get_by_test_id("operation-row").first
+    # Whole paths, both sides — nothing about the move is left to be inferred.
+    assert row.get_by_test_id("op-source").inner_text().endswith("/rome")
+    assert row.get_by_test_id("op-destination").inner_text().endswith(f"/{APPROVED}")
+    expect(row.get_by_test_id("op-type")).to_have_text("rename")
+    expect(row.get_by_test_id("op-count")).to_have_text("2")
+    expect(row.get_by_test_id("op-state")).to_have_text("planned")
+    assert errors == [], f"console errors: {errors}"
+
+
+def test_confirmation_shows_the_server_token_and_the_cancellation_limit(page, server):
+    _open_plan(page, server)
+    plan = httpx.get(f"{server.base}/api/v1/rename-plans", timeout=TIMEOUT).json()["items"][0]
+
+    token = page.get_by_test_id("confirm-token").inner_text()
+    assert plan["id"] in token and f"version {plan['version']}" in token
+    assert plan["checksum"][:12] in token
+    # The limit on cancellation is stated before the button, not discovered after.
+    assert "cannot be cancelled" in page.get_by_test_id("cancel-note").inner_text()
+
+
+def test_a_collision_is_previewed_as_blocking_and_cannot_be_applied(page, server):
+    _approve(server.base)
+    # Someone else already owns the destination name, with content worth keeping.
+    (server.seeded.lib / APPROVED).mkdir()
+    (server.seeded.lib / APPROVED / "precious.jpg").write_bytes(b"do not lose me")
+
+    page.goto(f"{server.base}/app/#/renames")
+    page.get_by_test_id("build-plan").click()
+    page.get_by_test_id("operations").wait_for()
+
+    expect(page.get_by_test_id("plan-state")).to_have_text("invalid")
+    expect(page.get_by_test_id("plan-blockers")).to_contain_text("destination_exists")
+    issue = page.get_by_test_id("op-issue").first
+    expect(issue).to_have_attribute("data-code", "destination_exists")
+    assert issue.inner_text().startswith("blocks")
+    expect(page.get_by_test_id("apply-plan")).to_be_disabled()
+
+    assert (server.seeded.lib / APPROVED / "precious.jpg").read_bytes() == b"do not lose me"
+    assert (server.seeded.lib / "rome").is_dir()
+
+
+# ── apply ────────────────────────────────────────────────────────────────────
+
+
+def test_apply_reports_progress_terminal_verification_and_moves_the_folder(page, server):
+    _open_plan(page, server)
+    # The progress region is live before anything runs, so the status it announces
+    # during the run reaches assistive technology.
+    progress = page.get_by_test_id("apply-progress")
+    expect(progress).to_have_attribute("aria-live", "polite")
+
+    page.get_by_test_id("apply-plan").click()
+
+    expect(page.get_by_test_id("apply-result")).to_contain_text("Applied 1, failed 0, skipped 0")
+    expect(page.get_by_test_id("plan-state")).to_have_text("applied")
+    row = page.get_by_test_id("operation-row").first
+    expect(row.get_by_test_id("op-state")).to_have_text("complete")
+    expect(row.get_by_test_id("op-verified")).to_be_visible()
+
+    assert (server.seeded.lib / APPROVED / "a.jpg").exists()
+    assert not (server.seeded.lib / "rome").exists()
+
+
+def test_a_stale_confirmation_is_refused_and_explained(page, server):
+    _open_plan(page, server)
+    # Another client applies first, which bumps the plan version this page is holding.
+    plan = httpx.get(f"{server.base}/api/v1/rename-plans", timeout=TIMEOUT).json()["items"][0]
+    httpx.post(
+        f"{server.base}/api/v1/rename-plans/{plan['id']}/apply",
+        json={"expected_version": plan["version"]},
+        timeout=TIMEOUT,
+    ).raise_for_status()
+
+    page.get_by_test_id("apply-plan").click()
+
+    expect(page.get_by_test_id("conflict")).to_be_visible()
+    # The refusal is not a second run: the page now shows the server's real state.
+    expect(page.get_by_test_id("plan-state")).to_have_text("applied")
+    expect(page.get_by_test_id("apply-plan")).to_be_disabled()
+
+
+def test_the_applied_plan_survives_a_reload(page, server):
+    _open_plan(page, server)
+    page.get_by_test_id("apply-plan").click()
+    expect(page.get_by_test_id("plan-state")).to_have_text("applied")
+
+    page.reload()
+    expect(page.get_by_test_id("plan-state")).to_have_text("applied")
+    expect(page.get_by_test_id("operation-row").first.get_by_test_id("op-state")).to_have_text(
+        "complete"
+    )
+    # The result banner is tab state, not server state, so it correctly does not return.
+    expect(page.get_by_test_id("apply-result")).to_have_count(0)
+
+
+def test_the_view_matches_the_journal_after_a_server_restart(page, server):
+    _open_plan(page, server)
+    page.get_by_test_id("apply-plan").click()
+    expect(page.get_by_test_id("plan-state")).to_have_text("applied")
+
+    server.stop()
+    server.start()  # same port, so the page reconnects to a genuinely fresh process
+
+    page.reload()
+    expect(page.get_by_test_id("plan-state")).to_have_text("applied")
+    expect(page.get_by_test_id("operation-row").first.get_by_test_id("op-state")).to_have_text(
+        "complete"
+    )
+    expect(page.get_by_test_id("op-verified")).to_be_visible()
+
+
+def test_apply_can_be_confirmed_from_the_keyboard(page, server):
+    _open_plan(page, server)
+    page.get_by_test_id("apply-plan").focus()
+    page.keyboard.press("Enter")
+    expect(page.get_by_test_id("plan-state")).to_have_text("applied")
+    assert (server.seeded.lib / APPROVED).is_dir()
+
+
+def test_a_finished_rename_offers_no_rollback(page, server):
+    """Rollback is recovery, not undo: once an operation is `complete` the journal
+    treats it as terminal, so the view must not suggest it can be reversed."""
+    _open_plan(page, server)
+    page.get_by_test_id("apply-plan").click()
+    expect(page.get_by_test_id("plan-state")).to_have_text("applied")
+    expect(page.get_by_test_id("rollback-plan")).to_have_count(0)
+
+
+def test_rollback_returns_an_unfinished_move_to_its_source(page, server):
+    _approve(server.base)
+    plan = _build(server.base)
+    # A crash after the move but before the database caught up: the content sits at
+    # the destination and the operation is still reversible.
+    _interrupt(server.seeded, plan, state="moving")
+    (server.seeded.lib / "rome").rename(server.seeded.lib / APPROVED)
+    with session_factory(server.seeded) as sf:
+        from photo_pipeline.services.rename_journal import RenameJournal
+
+        journal = RenameJournal(sf)
+        journal.transition(journal.operations(plan["id"])[0]["id"], "moved", fencing_token=1)
+
+    page.goto(f"{server.base}/app/#/renames")
+    page.get_by_test_id("rollback-plan").click()
+
+    expect(page.get_by_test_id("plan-state")).to_have_text("rolled_back")
+    assert (server.seeded.lib / "rome" / "a.jpg").exists()
+    assert not (server.seeded.lib / APPROVED).exists()
+
+
+# ── recovery ─────────────────────────────────────────────────────────────────
+
+
+def test_an_interrupted_rename_is_shown_and_offers_the_safe_action(page, server):
+    _approve(server.base)
+    plan = _build(server.base)
+    _interrupt(server.seeded, plan)  # source intact, destination free → resumable
+
+    page.goto(f"{server.base}/app/#/renames")
+    banner = page.get_by_test_id("recovery-banner")
+    banner.wait_for()
+    expect(page.get_by_test_id("recovery-item")).to_have_attribute(
+        "data-classification", "resumable"
+    )
+    # While it is unresolved nothing else may mutate the library.
+    expect(page.get_by_test_id("build-plan")).to_be_disabled()
+    expect(page.get_by_test_id("apply-plan")).to_be_disabled()
+
+    page.get_by_test_id("resolve-recovery").click()
+    expect(banner).to_have_count(0)
+    expect(page.get_by_test_id("build-plan")).to_be_enabled()
+
+
+def test_ambiguous_evidence_offers_no_automatic_action(page, server):
+    _approve(server.base)
+    plan = _build(server.base)
+    # Both paths exist: nothing can tell which side holds the truth.
+    _interrupt(server.seeded, plan, make_destination=True)
+
+    page.goto(f"{server.base}/app/#/renames")
+    page.get_by_test_id("recovery-banner").wait_for()
+    expect(page.get_by_test_id("recovery-item")).to_have_attribute("data-classification", "manual")
+    expect(page.get_by_test_id("manual-only")).to_be_visible()
+    # No button is offered for work the server would refuse to do.
+    expect(page.get_by_test_id("resolve-recovery")).to_have_count(0)
+
+
+def test_an_unresolved_rename_blocks_unrelated_name_changes(page, server):
+    _approve(server.base)
+    plan = _build(server.base)
+    _interrupt(server.seeded, plan, make_destination=True)
+
+    page.goto(f"{server.base}/app/#/albums?album=rome")
+    expect(page.get_by_test_id("rename-blocked")).to_be_visible()
+    expect(page.get_by_test_id("approve")).to_be_disabled()
+    expect(page.get_by_test_id("generate-proposals")).to_be_disabled()
+
+    # The disabled buttons are not the only guard: the API refuses too.
+    current = httpx.get(f"{server.base}/api/v1/albums/proposals/rome", timeout=TIMEOUT).json()
+    refused = httpx.post(
+        f"{server.base}/api/v1/albums/proposals/rome/approve",
+        json={"expected_version": current["version"]},
+        timeout=TIMEOUT,
+    )
+    assert refused.status_code == 409
+    assert refused.json()["error"]["code"] == "rename_recovery_required"

tests/e2e/test_traceability.py

@@ -12,12 +12,19 @@ from pathlib import Path
 REPO = Path(__file__).resolve().parents[2]
 MAP = json.loads((REPO / "tests" / "story_traceability.json").read_text())["stories"]
 PHASE_A_STORIES = {f"US01-0{n}" for n in range(1, 8)}
+PHASE_D_STORIES = {f"US04-0{n}" for n in range(1, 7)}
 
 
 def test_all_phase_a_stories_are_mapped():
     assert PHASE_A_STORIES <= set(MAP)
 
 
+def test_all_phase_d_stories_are_mapped():
+    """US04-06 acceptance: every guarded-rename story, plan through browser, is tied
+    to automated tests — renaming is the first thing that mutates the real library."""
+    assert PHASE_D_STORIES <= set(MAP)
+
+
 def test_every_mapped_test_file_exists_and_is_nonempty():
     for story, files in MAP.items():
         assert files, f"{story} maps to no tests"

tests/integration/test_rename_apply.py

@@ -0,0 +1,304 @@
+"""Applying and verifying guarded renames (US04-03).
+
+Everything here runs against a real temporary library: files really move, and the
+tests assert the bytes survived, the stable asset IDs kept their identity, the path
+history was recorded, and nothing unexpected was ever overwritten.
+"""
+
+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, AssetPath
+from photo_pipeline.services import hashing
+from photo_pipeline.services.rename_apply import (
+    ApplyConflict,
+    ApplyError,
+    RenameApplyService,
+)
+from photo_pipeline.services.rename_journal import JournalState, RenameJournal
+from photo_pipeline.services.renames import RenameService
+
+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, *, approved_name, names=("a.jpg", "b.jpg")):
+    folder = lib / album
+    folder.mkdir(parents=True, exist_ok=True)
+    with sf() as session:
+        for index, name in enumerate(names):
+            path = folder / name
+            path.write_bytes(f"content-{album}-{index}".encode())
+            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=hashing.sha256_file(path),
+                )
+            )
+        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 _plan(sf, lib):
+    return RenameService(sf, library_roots=(lib,)).build_plan()
+
+
+def _apply(sf, lib, plan, **kwargs):
+    service = RenameApplyService(sf, library_roots=(lib,))
+    return service.apply(plan["id"], expected_version=plan["version"], **kwargs)
+
+
+def _paths(sf):
+    with sf() as session:
+        return {a.id: a.current_path for a in session.scalars(select(Asset))}
+
+
+def _bytes_under(root):
+    return {
+        str(p.relative_to(root)): p.read_bytes() for p in sorted(root.rglob("*")) if p.is_file()
+    }
+
+
+# ── normal apply ─────────────────────────────────────────────────────────────
+
+
+def test_apply_moves_the_folder_and_preserves_bytes_and_identity(tmp_path):
+    _, sf, lib = _factory(tmp_path)
+    _album(sf, lib, "rome", approved_name="2019 Rome")
+    before_ids = set(_paths(sf))
+    before_bytes = _bytes_under(lib)
+
+    plan = _plan(sf, lib)
+    result = _apply(sf, lib, plan)
+
+    assert result["applied"] == 1 and result["failed"] == 0
+    assert result["state"] == "applied"
+    assert (lib / "2019 Rome").is_dir() and not (lib / "rome").exists()
+    # Same asset IDs, new paths, identical bytes.
+    after = _paths(sf)
+    assert set(after) == before_ids, "renaming must not change asset identity"
+    assert all("2019 Rome" in path for path in after.values())
+    assert _bytes_under(lib) == {
+        key.replace("rome/", "2019 Rome/"): value for key, value in before_bytes.items()
+    }
+
+
+def test_apply_records_the_path_history(tmp_path):
+    _, sf, lib = _factory(tmp_path)
+    _album(sf, lib, "rome", approved_name="2019 Rome", names=("a.jpg",))
+    plan = _plan(sf, lib)
+    _apply(sf, lib, plan)
+
+    with sf() as session:
+        rows = list(session.scalars(select(AssetPath).order_by(AssetPath.valid_from)))
+    open_rows = [r for r in rows if r.valid_until is None]
+    assert len(open_rows) == 1 and "2019 Rome" in open_rows[0].path
+    assert open_rows[0].reason == "rename"
+
+
+def test_every_operation_reaches_complete_in_the_journal(tmp_path):
+    _, sf, lib = _factory(tmp_path)
+    _album(sf, lib, "rome", approved_name="2019 Rome", names=("a.jpg",))
+    plan = _plan(sf, lib)
+    _apply(sf, lib, plan)
+
+    states = {op["journal_state"] for op in RenameJournal(sf).operations(plan["id"])}
+    assert states == {JournalState.COMPLETE}
+
+
+# ── confirmation and leases ──────────────────────────────────────────────────
+
+
+def test_apply_rejects_a_stale_version_without_touching_disk(tmp_path):
+    _, sf, lib = _factory(tmp_path)
+    _album(sf, lib, "rome", approved_name="2019 Rome")
+    plan = _plan(sf, lib)
+    before = _bytes_under(lib)
+
+    service = RenameApplyService(sf, library_roots=(lib,))
+    with pytest.raises(ApplyConflict):
+        service.apply(plan["id"], expected_version=plan["version"] + 5)
+    assert _bytes_under(lib) == before and (lib / "rome").exists()
+
+
+def test_apply_rejects_a_changed_checksum(tmp_path):
+    _, sf, lib = _factory(tmp_path)
+    _album(sf, lib, "rome", approved_name="2019 Rome")
+    plan = _plan(sf, lib)
+    service = RenameApplyService(sf, library_roots=(lib,))
+    with pytest.raises(ApplyConflict):
+        service.apply(
+            plan["id"], expected_version=plan["version"], expected_checksum="not-the-checksum"
+        )
+
+
+def test_an_invalid_plan_can_never_be_applied(tmp_path):
+    _, sf, lib = _factory(tmp_path)
+    _album(sf, lib, "rome", approved_name="2019 Rome")
+    (lib / "2019 Rome").mkdir()  # destination occupied → plan is invalid
+    plan = _plan(sf, lib)
+    assert plan["state"] == "invalid"
+
+    service = RenameApplyService(sf, library_roots=(lib,))
+    with pytest.raises(ApplyError):
+        service.apply(plan["id"], expected_version=plan["version"])
+
+
+def test_repeated_apply_is_a_no_op(tmp_path):
+    _, sf, lib = _factory(tmp_path)
+    _album(sf, lib, "rome", approved_name="2019 Rome")
+    plan = _plan(sf, lib)
+    _apply(sf, lib, plan)
+    after_first = _bytes_under(lib)
+
+    service = RenameApplyService(sf, library_roots=(lib,))
+    fresh_version = RenameService(sf, library_roots=(lib,)).get(plan["id"])["version"]
+    second = service.apply(plan["id"], expected_version=fresh_version)
+    assert second["applied"] == 0 and second["skipped"] == 1
+    assert _bytes_under(lib) == after_first
+
+
+# ── preconditions rechecked at mutation time ─────────────────────────────────
+
+
+def test_a_source_changed_after_planning_is_refused(tmp_path):
+    _, sf, lib = _factory(tmp_path)
+    folder = _album(sf, lib, "rome", approved_name="2019 Rome")
+    plan = _plan(sf, lib)
+    # The file changes between preview and confirmation.
+    (folder / "a.jpg").write_bytes(b"tampered")
+
+    result = _apply(sf, lib, plan)
+    assert result["failed"] == 1 and result["applied"] == 0
+    assert (lib / "rome").exists(), "a failed precondition must leave the source alone"
+    operation = RenameJournal(sf).operations(plan["id"])[0]
+    assert operation["journal_state"] == JournalState.FAILED
+    assert operation["error_code"] == "source_changed"
+
+
+def test_an_occupant_appearing_after_planning_is_never_overwritten(tmp_path):
+    _, sf, lib = _factory(tmp_path)
+    _album(sf, lib, "rome", approved_name="2019 Rome")
+    plan = _plan(sf, lib)
+    # Someone creates the destination between validation and apply.
+    (lib / "2019 Rome").mkdir()
+    (lib / "2019 Rome" / "precious.jpg").write_bytes(b"do not lose me")
+
+    result = _apply(sf, lib, plan)
+    assert result["failed"] == 1
+    assert (lib / "2019 Rome" / "precious.jpg").read_bytes() == b"do not lose me"
+    assert (lib / "rome").exists()
+
+
+# ── staged procedures ────────────────────────────────────────────────────────
+
+
+def test_case_only_rename_uses_the_staged_procedure(tmp_path):
+    _, sf, lib = _factory(tmp_path)
+    _album(sf, lib, "rome", approved_name="Rome", names=("a.jpg",))
+    plan = _plan(sf, lib)
+    assert plan["operations"][0]["case_only"] is True
+
+    result = _apply(sf, lib, plan)
+    assert result["applied"] == 1
+    # The directory entry now carries the new casing, and no staging dir survives.
+    entries = {p.name for p in lib.iterdir()}
+    assert "Rome" in entries
+    assert not any(name.startswith(".rename-") for name in entries)
+    assert list(_paths(sf).values())[0].endswith("Rome/a.jpg")
+
+
+def test_cross_filesystem_move_copies_verifies_then_deletes(tmp_path):
+    """The cross-filesystem branch is driven directly: a real second filesystem is
+    not portable in CI, but the copy-verify-delete procedure is what matters."""
+    _, sf, lib = _factory(tmp_path)
+    source = lib / "rome"
+    source.mkdir()
+    (source / "a.jpg").write_bytes(b"payload")
+    destination = lib / "2019 Rome"
+
+    RenameApplyService(sf, library_roots=(lib,))._copy_verify_delete(source, destination)
+
+    assert not source.exists()
+    assert (destination / "a.jpg").read_bytes() == b"payload"
+    assert not any(p.name.startswith(".rename-") for p in lib.iterdir())
+
+
+# ── restart durability ───────────────────────────────────────────────────────
+
+
+def test_applied_state_survives_a_restart(tmp_path):
+    config, sf, lib = _factory(tmp_path)
+    _album(sf, lib, "rome", approved_name="2019 Rome", names=("a.jpg",))
+    plan = _plan(sf, lib)
+    _apply(sf, lib, plan)
+    expected = _paths(sf)
+
+    reopened = create_session_factory(create_db_engine(config.database_url))
+    assert _paths(reopened) == expected
+    states = {op["journal_state"] for op in RenameJournal(reopened).operations(plan["id"])}
+    assert states == {JournalState.COMPLETE}
+
+
+# ── API ──────────────────────────────────────────────────────────────────────
+
+
+def test_api_apply_requires_the_current_version(tmp_path):
+    config, sf, lib = _factory(tmp_path)
+    _album(sf, lib, "rome", approved_name="2019 Rome", names=("a.jpg",))
+
+    with TestClient(create_app(config)) as client:
+        plan = client.post("/api/v1/rename-plans").json()
+
+        stale = client.post(
+            f"/api/v1/rename-plans/{plan['id']}/apply",
+            json={"expected_version": plan["version"] + 9},
+        )
+        assert stale.status_code == 409
+        assert stale.json()["error"]["code"] == "version_conflict"
+        assert (lib / "rome").exists()
+
+        applied = client.post(
+            f"/api/v1/rename-plans/{plan['id']}/apply",
+            json={"expected_version": plan["version"], "expected_checksum": plan["checksum"]},
+        ).json()
+        assert applied["applied"] == 1 and applied["state"] == "applied"
+        assert (lib / "2019 Rome").is_dir()
+
+        recovery = client.get("/api/v1/rename-recovery").json()
+        assert recovery["blocks_mutation"] is False and recovery["items"] == []

tests/integration/test_rename_recovery.py

@@ -0,0 +1,354 @@
+"""Recovering or rolling back interrupted renames (US04-04).
+
+The crash tests are real: a child process applies a plan and is killed by the
+``PHOTO_PIPELINE_FAULT_AFTER`` barrier at each persisted journal transition in turn.
+The parent then reopens the database, classifies the wreckage from journal + disk
+evidence, and asserts recovery converges without losing or overwriting anything.
+"""
+
+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 AlbumProposal, Asset
+from photo_pipeline.services import hashing
+from photo_pipeline.services.rename_apply import ApplyError, RenameApplyService
+from photo_pipeline.services.rename_journal import (
+    MANUAL,
+    RESUMABLE,
+    ROLLBACK_SAFE,
+    JournalState,
+)
+from photo_pipeline.services.renames import RenameService
+
+REPO = Path(__file__).resolve().parents[2]
+NOW = datetime(2026, 1, 1, tzinfo=timezone.utc)
+
+# A child process that applies the plan and dies at the configured barrier.
+APPLY_SCRIPT = """
+import sys
+sys.path.insert(0, {repo!r})
+from photo_pipeline.db import create_db_engine, create_session_factory
+from photo_pipeline.services.rename_apply import RenameApplyService
+
+db_url, lib, plan_id, version = sys.argv[1], sys.argv[2], sys.argv[3], int(sys.argv[4])
+sf = create_session_factory(create_db_engine(db_url))
+RenameApplyService(sf, library_roots=(lib,)).apply(plan_id, expected_version=version)
+"""
+
+
+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, *, approved_name, names=("a.jpg",)):
+    folder = lib / album
+    folder.mkdir(parents=True, exist_ok=True)
+    with sf() as session:
+        for index, name in enumerate(names):
+            path = folder / name
+            path.write_bytes(f"content-{album}-{index}".encode())
+            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=hashing.sha256_file(path),
+                )
+            )
+        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 _plan(sf, lib):
+    return RenameService(sf, library_roots=(lib,)).build_plan()
+
+
+def _crash_during_apply(config, lib, plan, barrier, tmp_path):
+    """Run apply in a child that dies right after ``barrier`` is persisted."""
+    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(lib),
+            plan["id"],
+            str(plan["version"]),
+        ],
+        env=env,
+        capture_output=True,
+    )
+    assert result.returncode == -9 or result.returncode == 9, (
+        f"child should have been killed at {barrier}, got {result.returncode}: "
+        f"{result.stderr.decode(errors='replace')[-400:]}"
+    )
+
+
+def _assets(sf):
+    with sf() as session:
+        return {a.id: a.current_path for a in session.scalars(select(Asset))}
+
+
+def _files(lib):
+    return {p.read_bytes() for p in lib.rglob("*") if p.is_file() and ".rename-" not in str(p)}
+
+
+# ── fault injection at every persisted transition ────────────────────────────
+
+
+@pytest.mark.parametrize(
+    "barrier",
+    [
+        JournalState.MOVING,
+        JournalState.MOVED,
+        JournalState.DATABASE_UPDATED,
+        JournalState.VERIFIED,
+    ],
+)
+def test_crash_at_each_transition_recovers_without_losing_content(barrier, tmp_path):
+    config, sf, lib = _factory(tmp_path)
+    _album(sf, lib, "rome", approved_name="2019 Rome")
+    plan = _plan(sf, lib)
+    content_before = _files(lib)
+
+    _crash_during_apply(config, lib, plan, barrier, tmp_path)
+
+    # Fresh process: the journal plus the disk decide what happened.
+    reopened = create_session_factory(create_db_engine(config.database_url))
+    service = RenameApplyService(reopened, library_roots=(lib,))
+    verdicts = service.journal.classify_all()
+    assert verdicts, f"a crash at {barrier} must leave recoverable work"
+    assert all(v["classification"] in {RESUMABLE, ROLLBACK_SAFE} for v in verdicts), verdicts
+
+    service.recover()
+
+    # No content was lost anywhere in the library.
+    assert _files(lib) == content_before
+    # And nothing is left half-done.
+    assert service.journal.blocks_mutation() is False
+
+
+@pytest.mark.parametrize(
+    "barrier", [JournalState.MOVING, JournalState.MOVED, JournalState.DATABASE_UPDATED]
+)
+def test_recovery_is_idempotent_across_repeated_restarts(barrier, tmp_path):
+    config, sf, lib = _factory(tmp_path)
+    _album(sf, lib, "rome", approved_name="2019 Rome")
+    plan = _plan(sf, lib)
+    _crash_during_apply(config, lib, plan, barrier, tmp_path)
+
+    reopened = create_session_factory(create_db_engine(config.database_url))
+    service = RenameApplyService(reopened, library_roots=(lib,))
+    service.recover()
+    settled_paths = _assets(reopened)
+    settled_files = _files(lib)
+
+    # Repeated recovery passes converge — the second and third change nothing.
+    for _ in range(2):
+        service.recover()
+        assert _assets(reopened) == settled_paths
+        assert _files(lib) == settled_files
+
+
+def test_crash_before_the_move_leaves_the_source_intact_and_resumable(tmp_path):
+    config, sf, lib = _factory(tmp_path)
+    _album(sf, lib, "rome", approved_name="2019 Rome")
+    plan = _plan(sf, lib)
+    _crash_during_apply(config, lib, plan, JournalState.MOVING, tmp_path)
+
+    reopened = create_session_factory(create_db_engine(config.database_url))
+    service = RenameApplyService(reopened, library_roots=(lib,))
+    verdict = service.journal.classify_all()[0]
+    assert verdict["classification"] == RESUMABLE
+    assert (lib / "rome").exists() and not (lib / "2019 Rome").exists()
+
+    service.recover()
+    # Resumable work returns to `planned`, so a fresh apply can finish the job.
+    operation = service.journal.operations(plan["id"])[0]
+    assert operation["journal_state"] == JournalState.PLANNED
+
+    fresh = RenameService(reopened, library_roots=(lib,)).get(plan["id"])
+    service.apply(plan["id"], expected_version=fresh["version"])
+    assert (lib / "2019 Rome").is_dir() and not (lib / "rome").exists()
+
+
+def test_crash_after_the_move_completes_the_bookkeeping(tmp_path):
+    config, sf, lib = _factory(tmp_path)
+    _album(sf, lib, "rome", approved_name="2019 Rome")
+    plan = _plan(sf, lib)
+    _crash_during_apply(config, lib, plan, JournalState.MOVED, tmp_path)
+
+    reopened = create_session_factory(create_db_engine(config.database_url))
+    service = RenameApplyService(reopened, library_roots=(lib,))
+    assert service.journal.classify_all()[0]["classification"] == ROLLBACK_SAFE
+
+    service.recover()
+    operation = service.journal.operations(plan["id"])[0]
+    assert operation["journal_state"] == JournalState.COMPLETE
+    # The database caught up with the filesystem.
+    assert all("2019 Rome" in path for path in _assets(reopened).values())
+
+
+# ── ambiguous evidence ───────────────────────────────────────────────────────
+
+
+def test_an_unexpected_destination_occupant_forces_manual_recovery(tmp_path):
+    config, sf, lib = _factory(tmp_path)
+    _album(sf, lib, "rome", approved_name="2019 Rome")
+    plan = _plan(sf, lib)
+    _crash_during_apply(config, lib, plan, JournalState.MOVING, tmp_path)
+
+    # Someone puts something at the destination while the app was down.
+    (lib / "2019 Rome").mkdir()
+    (lib / "2019 Rome" / "stranger.jpg").write_bytes(b"not ours")
+
+    reopened = create_session_factory(create_db_engine(config.database_url))
+    service = RenameApplyService(reopened, library_roots=(lib,))
+    verdict = service.journal.classify_all()[0]
+    assert verdict["classification"] == MANUAL
+    assert verdict["reason"]
+
+    result = service.recover()
+    assert result["manual"] == 1
+    # Nothing was touched, and the ambiguity keeps blocking other mutations.
+    assert (lib / "2019 Rome" / "stranger.jpg").read_bytes() == b"not ours"
+    assert (lib / "rome").exists()
+    assert service.journal.blocks_mutation() is True
+
+
+def test_manual_work_blocks_a_different_plan_from_applying(tmp_path):
+    config, sf, lib = _factory(tmp_path)
+    _album(sf, lib, "rome", approved_name="2019 Rome")
+    plan = _plan(sf, lib)
+    _crash_during_apply(config, lib, plan, JournalState.MOVING, tmp_path)
+    (lib / "2019 Rome").mkdir()  # ambiguous
+
+    reopened = create_session_factory(create_db_engine(config.database_url))
+    # The user parks the stuck album (un-approves it) and tries a different one, so
+    # the new plan is valid on its own — only the unresolved journal entry stands in
+    # the way.
+    with reopened() as session:
+        stuck = session.scalar(select(AlbumProposal).where(AlbumProposal.album == "rome"))
+        stuck.status = "edited"
+        session.commit()
+    _album(reopened, lib, "paris", approved_name="2020 Paris")
+    other = RenameService(reopened, library_roots=(lib,)).build_plan()
+    assert other["state"] == "validated", "the unrelated plan itself must be applyable"
+
+    service = RenameApplyService(reopened, library_roots=(lib,))
+    with pytest.raises(ApplyError, match="another rename is unresolved"):
+        service.apply(other["id"], expected_version=other["version"])
+    # Nothing moved for the unrelated album either.
+    assert (lib / "paris").exists() and not (lib / "2020 Paris").exists()
+
+
+# ── rollback ─────────────────────────────────────────────────────────────────
+
+
+def test_rollback_returns_the_content_to_its_source(tmp_path):
+    config, sf, lib = _factory(tmp_path)
+    _album(sf, lib, "rome", approved_name="2019 Rome")
+    plan = _plan(sf, lib)
+    before = _files(lib)
+    _crash_during_apply(config, lib, plan, JournalState.MOVED, tmp_path)
+
+    reopened = create_session_factory(create_db_engine(config.database_url))
+    service = RenameApplyService(reopened, library_roots=(lib,))
+    operation = service.journal.operations(plan["id"])[0]
+
+    service.rollback_operation(operation["id"])
+
+    assert (lib / "rome").is_dir() and not (lib / "2019 Rome").exists()
+    assert _files(lib) == before
+    assert service.journal.get(operation["id"])["journal_state"] == JournalState.ROLLED_BACK
+    assert all("rome/" in path for path in _assets(reopened).values())
+
+
+def test_rollback_refuses_when_the_source_is_occupied(tmp_path):
+    config, sf, lib = _factory(tmp_path)
+    _album(sf, lib, "rome", approved_name="2019 Rome")
+    plan = _plan(sf, lib)
+    _crash_during_apply(config, lib, plan, JournalState.MOVED, tmp_path)
+
+    # Something new appears at the old location while the app was down.
+    (lib / "rome").mkdir()
+    (lib / "rome" / "new.jpg").write_bytes(b"newcomer")
+
+    reopened = create_session_factory(create_db_engine(config.database_url))
+    service = RenameApplyService(reopened, library_roots=(lib,))
+    operation = service.journal.operations(plan["id"])[0]
+
+    with pytest.raises(ApplyError, match="occupied"):
+        service.rollback_operation(operation["id"])
+    assert (lib / "rome" / "new.jpg").read_bytes() == b"newcomer"
+
+
+def test_rollback_refuses_content_whose_bytes_changed(tmp_path):
+    config, sf, lib = _factory(tmp_path)
+    _album(sf, lib, "rome", approved_name="2019 Rome")
+    plan = _plan(sf, lib)
+    _crash_during_apply(config, lib, plan, JournalState.MOVED, tmp_path)
+
+    # The moved content is edited before rollback is attempted.
+    (lib / "2019 Rome" / "a.jpg").write_bytes(b"edited after the move")
+
+    reopened = create_session_factory(create_db_engine(config.database_url))
+    service = RenameApplyService(reopened, library_roots=(lib,))
+    operation = service.journal.operations(plan["id"])[0]
+
+    with pytest.raises(ApplyError, match="manual recovery"):
+        service.rollback_operation(operation["id"])
+    assert (lib / "2019 Rome" / "a.jpg").read_bytes() == b"edited after the move"
+
+
+def test_rollback_is_idempotent(tmp_path):
+    config, sf, lib = _factory(tmp_path)
+    _album(sf, lib, "rome", approved_name="2019 Rome")
+    plan = _plan(sf, lib)
+    _crash_during_apply(config, lib, plan, JournalState.MOVED, tmp_path)
+
+    reopened = create_session_factory(create_db_engine(config.database_url))
+    service = RenameApplyService(reopened, library_roots=(lib,))
+    operation = service.journal.operations(plan["id"])[0]
+    service.rollback_operation(operation["id"])
+    settled = _files(lib)
+
+    # A second rollback has nothing left to do and refuses rather than moving again.
+    with pytest.raises(ApplyError):
+        service.rollback_operation(operation["id"])
+    assert _files(lib) == settled

tests/integration/test_upload_batches.py

@@ -0,0 +1,588 @@
+"""Upload batch orchestration (US05-02).
+
+The uploader is a real executable on disk driven through the real integration
+adapter and ``subprocess`` — never a mock — so argument construction, output
+bounding, credential privacy, and killing a running process are exercised the way
+production does them. Each fake uploader records the argv it was given into a
+side-channel file, which is what the argument/album-isolation assertions read.
+"""
+
+import json
+import os
+import signal
+import stat
+import threading
+import time
+import uuid
+from datetime import datetime, timezone
+from http.server import BaseHTTPRequestHandler, HTTPServer
+from pathlib import Path
+
+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 UPLOAD_BATCH
+from photo_pipeline.jobs.worker import Worker
+from photo_pipeline.models import AnalysisResult, Asset, SafetyReview, UploadBatch
+from photo_pipeline.services.hashing import sha1_file, sha256_file
+from photo_pipeline.services.jobs import JobService, JobState
+from photo_pipeline.services.upload_batches import (
+    BatchConflict,
+    BatchError,
+    BatchState,
+    ItemState,
+    UploadBatchService,
+)
+from photo_pipeline.services.uploads import UploadService
+
+NOW = datetime(2026, 1, 1, tzinfo=timezone.utc)
+SENTINEL_KEY = "immich-sentinel-9f3a2b"
+UPLOADER_VERSION = "immich-go 0.21.0"
+
+
+# ── fake external boundary ───────────────────────────────────────────────────
+
+
+class _PingHandler(BaseHTTPRequestHandler):
+    def do_GET(self):  # noqa: N802 (BaseHTTPRequestHandler API)
+        self.send_response(200)
+        self.send_header("Content-Type", "application/json")
+        self.end_headers()
+        self.wfile.write(b'{"res":"pong"}')
+
+    def log_message(self, *args):
+        pass
+
+
+@pytest.fixture
+def immich_server():
+    server = HTTPServer(("127.0.0.1", 0), _PingHandler)
+    threading.Thread(target=server.serve_forever, daemon=True).start()
+    yield f"http://127.0.0.1:{server.server_port}"
+    server.shutdown()
+    server.server_close()
+
+
+def _uploader(tmp_path, body: str = "", *, name="immich-go"):
+    """A real executable standing in for immich-go.
+
+    ``--version`` answers like the real tool; any other invocation appends its
+    complete argv to ``<name>.argv`` and then runs ``body``.
+    """
+    path = tmp_path / name
+    argv_log = tmp_path / f"{name}.argv"
+    path.write_text(
+        "#!/bin/sh\n"
+        f'if [ "$1" = "--version" ]; then echo "{UPLOADER_VERSION}"; exit 0; fi\n'
+        f'printf "%s\\n" "$*" >> "{argv_log}"\n'
+        f"{body}\n"
+    )
+    path.chmod(path.stat().st_mode | stat.S_IEXEC | stat.S_IXGRP | stat.S_IXOTH)
+    return path
+
+
+def _argv(tmp_path, name="immich-go") -> list[str]:
+    log = tmp_path / f"{name}.argv"
+    return log.read_text().splitlines() if log.exists() else []
+
+
+# ── environment ──────────────────────────────────────────────────────────────
+
+
+def _env(tmp_path, server_url, *, uploader=None):
+    (tmp_path / "data").mkdir(exist_ok=True)
+    lib = tmp_path / "lib"
+    lib.mkdir(exist_ok=True)
+    config = Config.from_env(
+        {
+            "PHOTO_PIPELINE_DATA_DIR": str(tmp_path / "data"),
+            "PHOTO_PIPELINE_LIBRARY_ROOTS": str(lib),
+            "PHOTO_PIPELINE_IMMICH_SERVER_URL": server_url,
+            "PHOTO_PIPELINE_IMMICH_API_KEY": SENTINEL_KEY,
+            "PHOTO_PIPELINE_IMMICH_GO_BINARY": str(
+                uploader if uploader is not None else _uploader(tmp_path, "exit 0")
+            ),
+        }
+    )
+    run_migrations(config.database_url)
+    return config, create_session_factory(create_db_engine(config.database_url)), lib
+
+
+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:
+        for name in names:
+            path = folder / name
+            path.write_bytes(f"{album}/{name}".encode() * 16)
+            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(
+                SafetyReview(
+                    id=str(uuid.uuid4()),
+                    asset_id=asset_id,
+                    decision="sfw",
+                    exif_verified_at=NOW,
+                )
+            )
+            session.add(
+                AnalysisResult(asset_id=asset_id, status="analyzed", exif_written_at=NOW)
+            )
+        session.commit()
+    return folder
+
+
+def _service(sf, config):
+    return UploadBatchService(sf, config=config)
+
+
+def _approved(sf, config, albums=None, **kwargs):
+    """Create batches from a fresh, ready preflight."""
+    report = UploadService(sf, config=config).preflight(albums, **kwargs)
+    assert report["state"] == "ready", report["blockers"]
+    return _service(sf, config).create(albums, token=report["token"], **kwargs)
+
+
+# ── creation ─────────────────────────────────────────────────────────────────
+
+
+def test_batch_records_album_assets_hashes_and_command(tmp_path, immich_server):
+    config, sf, lib = _env(tmp_path, immich_server)
+    folder = _album(sf, lib)
+
+    (batch,) = _approved(sf, config)
+
+    assert batch["state"] == BatchState.PLANNED
+    assert batch["album"] == "rome" and batch["folder"] == str(folder)
+    assert batch["asset_count"] == 2 and len(batch["items"]) == 2
+    assert batch["uploader_version"] == UPLOADER_VERSION
+    assert "--album-name=rome" in batch["command"]
+    assert "--api-key=***" in batch["command"]
+    for item in batch["items"]:
+        assert item["sha256"] == sha256_file(item["path"])
+        assert item["sha1"] == sha1_file(item["path"])
+        assert item["state"] == ItemState.PENDING
+
+
+def test_one_batch_per_album_and_scope_is_isolated(tmp_path, immich_server):
+    config, sf, lib = _env(tmp_path, immich_server)
+    _album(sf, lib, "rome")
+    _album(sf, lib, "paris", names=("c.jpg",))
+
+    batches = _approved(sf, config, ["paris"])
+
+    assert [b["album"] for b in batches] == ["paris"]
+    assert [b["asset_count"] for b in batches] == [1]
+
+
+def test_creating_twice_reuses_the_open_batch(tmp_path, immich_server):
+    config, sf, lib = _env(tmp_path, immich_server)
+    _album(sf, lib)
+
+    first = _approved(sf, config)[0]
+    second = _approved(sf, config)[0]
+
+    assert first["id"] == second["id"]
+
+
+def test_stale_preflight_token_cannot_create_a_batch(tmp_path, immich_server):
+    config, sf, lib = _env(tmp_path, immich_server)
+    folder = _album(sf, lib)
+    token = UploadService(sf, config=config).preflight()["token"]
+    (folder / "a.jpg").write_bytes(b"edited after approval")
+
+    with pytest.raises(BatchConflict) as error:
+        _service(sf, config).create(token=token)
+
+    assert error.value.code == "stale_preflight"
+
+
+def test_blocked_scope_cannot_create_a_batch(tmp_path, immich_server):
+    config, sf, lib = _env(tmp_path, immich_server)
+    folder = _album(sf, lib)
+    (folder / "a.jpg").write_bytes(b"edited before approval")
+    report = UploadService(sf, config=config).preflight()
+
+    with pytest.raises(BatchError) as error:
+        _service(sf, config).create(token=report["token"])
+
+    assert error.value.code == "not_ready"
+
+
+# ── running the uploader ─────────────────────────────────────────────────────
+
+
+def test_run_invokes_the_uploader_with_the_batch_album_and_folder(tmp_path, immich_server):
+    config, sf, lib = _env(tmp_path, immich_server)
+    folder = _album(sf, lib)
+    (batch,) = _approved(sf, config)
+
+    result = _service(sf, config).run(batch["id"])
+
+    assert result["state"] == BatchState.SUCCEEDED and result["exit_code"] == 0
+    (invocation,) = _argv(tmp_path)
+    assert "upload from-folder" in invocation
+    assert "--album-name=rome" in invocation
+    assert str(folder) in invocation
+    assert all(item["state"] == ItemState.SENT for item in result["items"])
+    assert result["attempt_count"] == 1 and result["started_at"] and result["finished_at"]
+
+
+def test_other_albums_are_never_passed_to_the_uploader(tmp_path, immich_server):
+    config, sf, lib = _env(tmp_path, immich_server)
+    _album(sf, lib, "rome")
+    paris = _album(sf, lib, "paris", names=("c.jpg",))
+    (batch,) = _approved(sf, config, ["rome"])
+
+    _service(sf, config).run(batch["id"])
+
+    (invocation,) = _argv(tmp_path)
+    assert str(paris) not in invocation and "paris" not in invocation
+
+
+def test_album_names_are_arguments_not_shell_text(tmp_path, immich_server):
+    """A folder whose name contains shell metacharacters must reach the uploader
+    verbatim; nothing may be interpreted (no shell is involved)."""
+    config, sf, lib = _env(tmp_path, immich_server)
+    hostile = "rome; touch pwned"
+    _album(sf, lib, hostile, names=("a.jpg",))
+    (batch,) = _approved(sf, config)
+
+    result = _service(sf, config).run(batch["id"])
+
+    assert result["state"] == BatchState.SUCCEEDED
+    assert f"--album-name={hostile}" in _argv(tmp_path)[0]
+    assert not (Path.cwd() / "pwned").exists() and not (lib / "pwned").exists()
+
+
+def test_uploader_failure_is_recorded_and_items_are_not_sent(tmp_path, immich_server):
+    config, sf, lib = _env(
+        tmp_path, immich_server, uploader=_uploader(tmp_path, "echo 'boom' >&2; exit 3")
+    )
+    _album(sf, lib)
+    (batch,) = _approved(sf, config)
+
+    result = _service(sf, config).run(batch["id"])
+
+    assert result["state"] == BatchState.FAILED and result["exit_code"] == 3
+    assert result["error_code"] == "uploader_failed"
+    assert all(item["state"] == ItemState.FAILED for item in result["items"])
+    assert "boom" in _service(sf, config).report(batch["id"])
+
+
+def test_a_failed_batch_can_be_retried_as_a_new_attempt(tmp_path, immich_server):
+    config, sf, lib = _env(tmp_path, immich_server, uploader=_uploader(tmp_path, "exit 3"))
+    _album(sf, lib)
+    (batch,) = _approved(sf, config)
+    service = _service(sf, config)
+    service.run(batch["id"])
+
+    retried = service.run(batch["id"])
+
+    assert retried["attempt_count"] == 2
+    assert len(_argv(tmp_path)) == 2
+
+
+def test_bytes_edited_after_approval_block_the_upload(tmp_path, immich_server):
+    config, sf, lib = _env(tmp_path, immich_server)
+    folder = _album(sf, lib)
+    (batch,) = _approved(sf, config)
+    (folder / "a.jpg").write_bytes(b"edited after approval")
+
+    with pytest.raises(BatchConflict) as error:
+        _service(sf, config).run(batch["id"])
+
+    assert error.value.code == "stale_preflight"
+    assert _argv(tmp_path) == [], "the uploader must not run on changed bytes"
+    assert _service(sf, config).get(batch["id"])["state"] == BatchState.FAILED
+
+
+# ── credential privacy ───────────────────────────────────────────────────────
+
+
+def test_the_api_key_reaches_the_uploader_but_never_the_record(tmp_path, immich_server):
+    config, sf, lib = _env(tmp_path, immich_server)
+    _album(sf, lib)
+    (batch,) = _approved(sf, config)
+
+    result = _service(sf, config).run(batch["id"])
+
+    assert f"--api-key={SENTINEL_KEY}" in _argv(tmp_path)[0], "the real key must be passed"
+    assert SENTINEL_KEY not in json.dumps(result, default=str)
+
+
+def test_a_key_echoed_by_the_uploader_is_scrubbed_from_the_report(tmp_path, immich_server):
+    config, sf, lib = _env(
+        tmp_path, immich_server, uploader=_uploader(tmp_path, 'echo "using key $4"; exit 0')
+    )
+    _album(sf, lib)
+    (batch,) = _approved(sf, config)
+    service = _service(sf, config)
+
+    service.run(batch["id"])
+
+    report = service.report(batch["id"])
+    assert SENTINEL_KEY not in report and "***" in report
+
+
+# ── bounded output ───────────────────────────────────────────────────────────
+
+
+def test_a_chatty_uploader_cannot_grow_the_report_without_bound(tmp_path, immich_server):
+    config, sf, lib = _env(
+        tmp_path,
+        immich_server,
+        uploader=_uploader(tmp_path, "i=0; while [ $i -lt 2000 ]; do echo line-$i; i=$((i+1)); done"),
+    )
+    _album(sf, lib)
+    (batch,) = _approved(sf, config)
+
+    from photo_pipeline.integrations import immich_go
+
+    service = _service(sf, config)
+    original = immich_go.MAX_REPORT_BYTES
+    immich_go.MAX_REPORT_BYTES = 500  # a cap small enough to hit in one test
+    try:
+        result = service.run(batch["id"])
+    finally:
+        immich_go.MAX_REPORT_BYTES = original
+
+    assert result["state"] == BatchState.SUCCEEDED, "the uploader still finished normally"
+    assert result["report_truncated"] is True
+    assert result["report_bytes"] <= 500
+    assert Path(result["report_path"]).stat().st_size <= 500
+
+
+# ── one lane ─────────────────────────────────────────────────────────────────
+
+
+def test_a_second_batch_cannot_run_while_one_is_running(tmp_path, immich_server):
+    config, sf, lib = _env(tmp_path, immich_server)
+    _album(sf, lib, "rome")
+    _album(sf, lib, "paris", names=("c.jpg",))
+    rome, paris = sorted(_approved(sf, config), key=lambda b: b["album"] != "rome")
+    with sf() as session:  # a batch already occupying the lane
+        session.get(UploadBatch, rome["id"]).state = BatchState.RUNNING
+        session.commit()
+
+    with pytest.raises(BatchConflict) as error:
+        _service(sf, config).run(paris["id"])
+
+    assert error.value.code == "lane_busy"
+    assert _argv(tmp_path) == []
+
+
+def test_the_job_lock_refuses_a_second_queued_upload(tmp_path, immich_server):
+    config, sf, lib = _env(tmp_path, immich_server)
+    _album(sf, lib, "rome")
+    _album(sf, lib, "paris", names=("c.jpg",))
+    batches = _approved(sf, config)
+
+    with TestClient(create_app(config)) as client:
+        first = client.post(f"/api/v1/upload-batches/{batches[0]['id']}/start")
+        second = client.post(f"/api/v1/upload-batches/{batches[1]['id']}/start")
+
+    assert first.status_code == 200
+    assert second.status_code == 409 and second.json()["error"]["code"] == "lock_held"
+
+
+# ── cancellation ─────────────────────────────────────────────────────────────
+
+
+def test_cancelling_a_running_batch_stops_the_uploader(tmp_path, immich_server):
+    """The uploader sleeps; cancelling flips the batch to ``cancelling`` and the
+    running attempt must terminate the process and record ``cancelled``."""
+    config, sf, lib = _env(
+        tmp_path, immich_server, uploader=_uploader(tmp_path, 'echo started; sleep 30; exit 0')
+    )
+    _album(sf, lib)
+    (batch,) = _approved(sf, config)
+    service = _service(sf, config)
+    outcome = {}
+
+    def _run():
+        outcome["batch"] = service.run(batch["id"])
+
+    runner = threading.Thread(target=_run)
+    started = time.monotonic()
+    runner.start()
+    while service.get(batch["id"])["state"] != BatchState.RUNNING:
+        assert time.monotonic() - started < 30, "the attempt never started"
+        time.sleep(0.02)
+
+    service.cancel(batch["id"])
+    runner.join(timeout=30)
+
+    assert not runner.is_alive(), "cancellation must not wait for the uploader's own timeout"
+    assert outcome["batch"]["state"] == BatchState.CANCELLED
+    assert all(item["state"] == ItemState.PENDING for item in outcome["batch"]["items"])
+
+
+def test_cancelling_a_planned_batch_never_starts_the_uploader(tmp_path, immich_server):
+    config, sf, lib = _env(tmp_path, immich_server)
+    _album(sf, lib)
+    (batch,) = _approved(sf, config)
+    service = _service(sf, config)
+
+    cancelled = service.cancel(batch["id"])
+
+    assert cancelled["state"] == BatchState.CANCELLED
+    assert _argv(tmp_path) == []
+
+
+def test_a_cancelled_batch_can_be_run_again(tmp_path, immich_server):
+    config, sf, lib = _env(tmp_path, immich_server)
+    _album(sf, lib)
+    (batch,) = _approved(sf, config)
+    service = _service(sf, config)
+    service.cancel(batch["id"])
+
+    result = service.run(batch["id"])
+
+    assert result["state"] == BatchState.SUCCEEDED and result["attempt_count"] == 1
+
+
+# ── restart / recovery ───────────────────────────────────────────────────────
+
+
+def test_an_interrupted_attempt_becomes_uncertain_not_failed(tmp_path, immich_server):
+    """The process died mid-upload: Immich may hold the files, so the batch requires
+    verification (US05-04) instead of a blind retry, and the lane is released."""
+    config, sf, lib = _env(tmp_path, immich_server)
+    _album(sf, lib)
+    (batch,) = _approved(sf, config)
+    with sf() as session:  # what a killed worker leaves behind
+        session.get(UploadBatch, batch["id"]).state = BatchState.RUNNING
+        session.commit()
+    service = _service(sf, config)
+
+    assert service.recover() == {"interrupted": 1}
+
+    recovered = service.get(batch["id"])
+    assert recovered["state"] == BatchState.UNKNOWN
+    assert recovered["error_code"] == "interrupted"
+    with pytest.raises(BatchError) as error:
+        service.run(batch["id"])
+    assert error.value.code == "requires_verification"
+    assert _argv(tmp_path) == []
+
+
+def test_recovery_leaves_a_planned_batch_runnable(tmp_path, immich_server):
+    config, sf, lib = _env(tmp_path, immich_server)
+    _album(sf, lib)
+    (batch,) = _approved(sf, config)
+    service = _service(sf, config)
+
+    assert service.recover() == {"interrupted": 0}
+
+    assert service.run(batch["id"])["state"] == BatchState.SUCCEEDED
+
+
+def test_application_startup_recovers_an_interrupted_batch(tmp_path, immich_server):
+    config, sf, lib = _env(tmp_path, immich_server)
+    _album(sf, lib)
+    (batch,) = _approved(sf, config)
+    with sf() as session:
+        session.get(UploadBatch, batch["id"]).state = BatchState.RUNNING
+        session.commit()
+
+    with TestClient(create_app(config)) as client:
+        state = client.get(f"/api/v1/upload-batches/{batch['id']}").json()["state"]
+
+    assert state == BatchState.UNKNOWN
+
+
+def test_a_killed_uploader_leaves_an_uncertain_batch(tmp_path, immich_server):
+    """A real SIGKILL of the uploader process, not a simulated state write."""
+    config, sf, lib = _env(
+        tmp_path,
+        immich_server,
+        uploader=_uploader(tmp_path, 'echo "pid $$"; sleep 30; exit 0'),
+    )
+    _album(sf, lib)
+    (batch,) = _approved(sf, config)
+    service = _service(sf, config)
+    outcome = {}
+    runner = threading.Thread(target=lambda: outcome.update(batch=service.run(batch["id"])))
+    runner.start()
+
+    report = tmp_path / "data" / "uploads"
+    deadline = time.monotonic() + 30
+    pid = None
+    while pid is None:
+        assert time.monotonic() < deadline, "the uploader never announced itself"
+        for log in report.glob("*.log"):
+            text = log.read_text()
+            if text.startswith("pid "):
+                pid = int(text.split()[1])
+        time.sleep(0.02)
+    os.kill(pid, signal.SIGKILL)
+    runner.join(timeout=30)
+
+    # The process is gone with a non-zero status and no parsed report: the attempt
+    # failed locally, and a restart classifies it honestly.
+    assert outcome["batch"]["state"] in (BatchState.FAILED, BatchState.UNKNOWN)
+    assert outcome["batch"]["exit_code"] != 0
+
+
+# ── worker integration ───────────────────────────────────────────────────────
+
+
+def test_the_worker_runs_a_queued_batch_on_the_upload_lane(tmp_path, immich_server):
+    config, sf, lib = _env(tmp_path, immich_server)
+    _album(sf, lib)
+    (batch,) = _approved(sf, config)
+
+    with TestClient(create_app(config)) as client:
+        response = client.post(f"/api/v1/upload-batches/{batch['id']}/start")
+    job_id = response.json()["job"]["id"]
+    Worker(sf, worker_id="uploader-1", job_types=[UPLOAD_BATCH], config=config).run_once()
+
+    assert JobService(sf).get(job_id)["state"] == JobState.SUCCEEDED
+    assert _service(sf, config).get(batch["id"])["state"] == BatchState.SUCCEEDED
+    assert len(_argv(tmp_path)) == 1
+
+
+# ── API surface ──────────────────────────────────────────────────────────────
+
+
+def test_api_creates_lists_and_reads_batches_without_secrets(tmp_path, immich_server):
+    config, sf, lib = _env(tmp_path, immich_server)
+    _album(sf, lib)
+
+    with TestClient(create_app(config)) as client:
+        token = client.post("/api/v1/upload-preflight", json={}).json()["token"]
+        created = client.post("/api/v1/upload-batches", json={"token": token})
+        listed = client.get("/api/v1/upload-batches")
+        batch_id = created.json()["batches"][0]["id"]
+        fetched = client.get(f"/api/v1/upload-batches/{batch_id}")
+
+    assert created.status_code == 201
+    assert listed.json()["batches"][0]["id"] == batch_id
+    assert fetched.json()["state"] == BatchState.PLANNED
+    assert SENTINEL_KEY not in created.text + listed.text + fetched.text
+
+
+def test_api_rejects_a_stale_token_and_an_unknown_batch(tmp_path, immich_server):
+    config, sf, lib = _env(tmp_path, immich_server)
+    _album(sf, lib)
+
+    with TestClient(create_app(config)) as client:
+        stale = client.post("/api/v1/upload-batches", json={"token": "v1:not-the-token"})
+        missing = client.get("/api/v1/upload-batches/does-not-exist")
+
+    assert stale.status_code == 409 and stale.json()["error"]["code"] == "stale_preflight"
+    assert missing.status_code == 404

tests/integration/test_upload_preflight.py

@@ -0,0 +1,511 @@
+"""Upload preflight: credentials, scope, readiness, and tokens (US05-01).
+
+Every external boundary is faked but never mocked away: the uploader is a real
+executable on disk invoked through ``subprocess``, and the Immich server is a real
+localhost HTTP server answering ``/api/server/ping``. Preflight itself must stay
+read-only — the library snapshot is asserted unchanged.
+"""
+
+import json
+import stat
+import threading
+import uuid
+from datetime import datetime, timezone
+from http.server import BaseHTTPRequestHandler, HTTPServer
+
+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.models import (
+    AnalysisResult,
+    Asset,
+    RenameOperation,
+    RenamePlan,
+    SafetyReview,
+)
+from photo_pipeline.services.hashing import sha256_file
+from photo_pipeline.services.uploads import UploadError, UploadService
+
+NOW = datetime(2026, 1, 1, tzinfo=timezone.utc)
+# A sentinel credential: every assertion below proves it never leaves configuration.
+SENTINEL_KEY = "immich-sentinel-9f3a2b"
+UPLOADER_VERSION = "immich-go 0.21.0"
+
+
+# ── fake external boundary ───────────────────────────────────────────────────
+
+
+class _PingHandler(BaseHTTPRequestHandler):
+    payload = b'{"res":"pong"}'
+    status = 200
+
+    def do_GET(self):  # noqa: N802 (BaseHTTPRequestHandler API)
+        self.send_response(type(self).status)
+        self.send_header("Content-Type", "application/json")
+        self.end_headers()
+        self.wfile.write(type(self).payload)
+
+    def log_message(self, *args):
+        pass  # keep the test output clean
+
+
+@pytest.fixture
+def immich_server():
+    """A real HTTP server that answers like Immich. Yields its base URL."""
+    handler = type("Handler", (_PingHandler,), {})
+    server = HTTPServer(("127.0.0.1", 0), handler)
+    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    yield f"http://127.0.0.1:{server.server_port}", handler
+    server.shutdown()
+    server.server_close()
+
+
+def _fake_uploader(tmp_path):
+    """A real executable standing in for immich-go."""
+    path = tmp_path / "immich-go"
+    path.write_text(f"#!/bin/sh\necho '{UPLOADER_VERSION}'\n")
+    path.chmod(path.stat().st_mode | stat.S_IEXEC | stat.S_IXGRP | stat.S_IXOTH)
+    return path
+
+
+# ── environment ──────────────────────────────────────────────────────────────
+
+
+def _env(tmp_path, server_url, *, credential=SENTINEL_KEY, uploader=None):
+    (tmp_path / "data").mkdir(exist_ok=True)
+    lib = tmp_path / "lib"
+    lib.mkdir(exist_ok=True)
+    env = {
+        "PHOTO_PIPELINE_DATA_DIR": str(tmp_path / "data"),
+        "PHOTO_PIPELINE_LIBRARY_ROOTS": str(lib),
+        "PHOTO_PIPELINE_IMMICH_SERVER_URL": server_url,
+        "PHOTO_PIPELINE_IMMICH_GO_BINARY": str(
+            uploader if uploader is not None else _fake_uploader(tmp_path)
+        ),
+    }
+    if credential:
+        env["PHOTO_PIPELINE_IMMICH_API_KEY"] = credential
+    config = Config.from_env(env)
+    run_migrations(config.database_url)
+    return config, create_session_factory(create_db_engine(config.database_url)), lib
+
+
+def _album(
+    sf,
+    lib,
+    album="rome",
+    names=("a.jpg", "b.jpg"),
+    *,
+    decision="sfw",
+    exif_verified=True,
+    analyzed=True,
+):
+    """A real album folder with registered, stage-complete assets."""
+    folder = lib / album
+    folder.mkdir(parents=True, exist_ok=True)
+    ids = []
+    with sf() as session:
+        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 decision is not None:
+                session.add(
+                    SafetyReview(
+                        id=str(uuid.uuid4()),
+                        asset_id=asset_id,
+                        decision=decision,
+                        exif_verified_at=NOW if exif_verified else None,
+                    )
+                )
+            if analyzed:
+                session.add(
+                    AnalysisResult(
+                        asset_id=asset_id,
+                        status="analyzed",
+                        exif_written_at=NOW,
+                    )
+                )
+        session.commit()
+    return folder, ids
+
+
+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 _service(sf, config):
+    return UploadService(sf, config=config)
+
+
+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"]
+    }
+
+
+# ── happy path ───────────────────────────────────────────────────────────────
+
+
+def test_ready_preflight_reports_scope_command_and_hashes(tmp_path, immich_server):
+    url, _ = immich_server
+    config, sf, lib = _env(tmp_path, url)
+    folder, ids = _album(sf, lib)
+    before = _snapshot(lib)
+
+    report = _service(sf, config).preflight()
+
+    assert report["state"] == "ready" and report["blockers"] == []
+    assert report["server"]["reachable"] is True
+    assert report["uploader"]["installed"] is True
+    assert report["uploader"]["version"] == UPLOADER_VERSION
+    assert report["credentials"]["api_key_configured"] is True
+    assert report["totals"] == {
+        "albums": 1,
+        "ready_albums": 1,
+        "assets": 2,
+        "eligible": 2,
+        "blocked": 0,
+    }
+    album = report["albums"][0]
+    assert album["album"] == "rome" and album["folder"] == str(folder)
+    assert album["album_name"] == "rome"  # folder-as-album
+    assert album["partial"] is False and album["state"] == "ready"
+    assert sorted(a["asset_id"] for a in album["assets"]) == sorted(ids)
+    # Hashes are of the bytes on disk right now, not a remembered value.
+    for asset in album["assets"]:
+        assert asset["current_sha256"] == sha256_file(asset["current_path"])
+    assert report["token"].startswith("v1:")
+    assert _snapshot(lib) == before, "preflight must not touch the library"
+
+
+def test_command_preview_shows_folder_and_album_without_the_key(tmp_path, immich_server):
+    url, _ = immich_server
+    config, sf, lib = _env(tmp_path, url)
+    folder, _ = _album(sf, lib)
+
+    preview = _service(sf, config).preflight()["albums"][0]["command_preview"]
+
+    assert f"--api-key={'***'}" in preview
+    assert "--album-name=rome" in preview
+    assert str(folder) in preview
+    assert SENTINEL_KEY not in " ".join(preview)
+
+
+def test_scoping_to_one_album_excludes_the_others(tmp_path, immich_server):
+    url, _ = immich_server
+    config, sf, lib = _env(tmp_path, url)
+    _album(sf, lib, "rome")
+    _album(sf, lib, "paris", names=("c.jpg",))
+
+    report = _service(sf, config).preflight(["paris"])
+
+    assert [album["album"] for album in report["albums"]] == ["paris"]
+    assert report["totals"]["assets"] == 1
+
+
+def test_unknown_album_in_scope_is_refused(tmp_path, immich_server):
+    url, _ = immich_server
+    config, sf, lib = _env(tmp_path, url)
+    _album(sf, lib, "rome")
+
+    with pytest.raises(UploadError):
+        _service(sf, config).preflight(["atlantis"])
+
+
+def test_empty_scope_is_a_blocker(tmp_path, immich_server):
+    url, _ = immich_server
+    config, sf, _ = _env(tmp_path, url)
+
+    report = _service(sf, config).preflight()
+
+    assert report["state"] == "blocked" and "empty_scope" in _codes(report)
+
+
+# ── credentials and environment ──────────────────────────────────────────────
+
+
+def test_missing_api_key_blocks_without_touching_the_server(tmp_path, immich_server):
+    url, _ = immich_server
+    config, sf, lib = _env(tmp_path, url, credential=None)
+    _album(sf, lib)
+
+    report = _service(sf, config).preflight()
+
+    assert report["state"] == "blocked"
+    assert "credentials_missing" in _codes(report)
+    assert report["credentials"]["api_key_configured"] is False
+
+
+def test_unreachable_server_blocks(tmp_path, immich_server):
+    url, _ = immich_server
+    config, sf, lib = _env(tmp_path, url.replace(url.rsplit(":", 1)[-1], "1"))
+    _album(sf, lib)
+
+    report = _service(sf, config).preflight()
+
+    assert "server_unreachable" in _codes(report)
+    assert report["server"]["reachable"] is False and report["server"]["detail"]
+
+
+def test_server_that_is_not_immich_blocks(tmp_path, immich_server):
+    url, handler = immich_server
+    handler.payload = b'{"error":"unauthorized"}'
+    handler.status = 401
+    config, sf, lib = _env(tmp_path, url)
+    _album(sf, lib)
+
+    assert "server_unreachable" in _codes(_service(sf, config).preflight())
+
+
+def test_missing_uploader_blocks(tmp_path, immich_server):
+    url, _ = immich_server
+    config, sf, lib = _env(tmp_path, url, uploader=tmp_path / "does-not-exist")
+    _album(sf, lib)
+
+    report = _service(sf, config).preflight()
+
+    assert "immich_go_missing" in _codes(report)
+    assert report["uploader"]["installed"] is False and report["uploader"]["version"] is None
+
+
+def test_half_applied_rename_blocks_upload(tmp_path, immich_server):
+    """Album paths must be final before upload: an operation that may have already
+    touched the disk blocks every other mutation until it is recovered (US04-04)."""
+    url, _ = immich_server
+    config, sf, lib = _env(tmp_path, url)
+    folder, _ = _album(sf, lib)
+    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()
+
+    assert report["state"] == "blocked" and "rename_pending" in _codes(report)
+
+
+def test_no_secret_appears_anywhere_in_the_report(tmp_path, immich_server):
+    url, _ = immich_server
+    config, sf, lib = _env(tmp_path, url)
+    _album(sf, lib)
+
+    report = _service(sf, config).preflight()
+
+    assert SENTINEL_KEY not in json.dumps(report, default=str)
+
+
+# ── stage readiness ──────────────────────────────────────────────────────────
+
+
+@pytest.mark.parametrize(
+    "kwargs,code",
+    [
+        ({"decision": None}, "safety_undecided"),
+        ({"decision": "deferred"}, "safety_deferred"),
+        ({"exif_verified": False}, "safety_exif_unverified"),
+        ({"analyzed": False}, "analysis_incomplete"),
+    ],
+)
+def test_blocked_stage_blocks_upload(tmp_path, immich_server, kwargs, code):
+    url, _ = immich_server
+    config, sf, lib = _env(tmp_path, url)
+    _album(sf, lib, **kwargs)
+
+    report = _service(sf, config).preflight()
+
+    assert report["state"] == "blocked"
+    assert code in _codes(report)
+    assert report["albums"][0]["eligible_count"] == 0
+
+
+def test_reviewed_nsfw_asset_is_upload_eligible_without_analysis(tmp_path, immich_server):
+    """NSFW never reaches the analyser, but a verified nsfw keyword makes it
+    uploadable (concept §8 eligibility table)."""
+    url, _ = immich_server
+    config, sf, lib = _env(tmp_path, url)
+    _album(sf, lib, decision="nsfw", analyzed=False)
+
+    report = _service(sf, config).preflight()
+
+    assert report["state"] == "ready"
+    assert report["albums"][0]["eligible_count"] == 2
+
+
+def test_changed_bytes_block_the_album(tmp_path, immich_server):
+    url, _ = immich_server
+    config, sf, lib = _env(tmp_path, url)
+    folder, _ = _album(sf, lib)
+    (folder / "a.jpg").write_bytes(b"edited after the checkpoint")
+
+    report = _service(sf, config).preflight()
+
+    assert "bytes_changed" in _codes(report)
+    assert report["albums"][0]["blocked_count"] == 1
+
+
+def test_missing_file_blocks_the_album(tmp_path, immich_server):
+    url, _ = immich_server
+    config, sf, lib = _env(tmp_path, url)
+    folder, _ = _album(sf, lib)
+    (folder / "a.jpg").unlink()
+
+    assert "file_missing" in _codes(_service(sf, config).preflight())
+
+
+# ── partial scope ────────────────────────────────────────────────────────────
+
+
+def test_partial_album_is_blocked_until_explicitly_approved(tmp_path, immich_server):
+    url, _ = immich_server
+    config, sf, lib = _env(tmp_path, url)
+    folder, _ = _album(sf, lib)
+    (folder / "a.jpg").write_bytes(b"changed")
+
+    blocked = _service(sf, config).preflight()
+    approved = _service(sf, config).preflight(allow_partial=True)
+
+    assert blocked["state"] == "blocked"
+    assert "partial_scope" in {issue["code"] for issue in blocked["albums"][0]["blockers"]}
+    assert approved["state"] == "ready"
+    assert approved["albums"][0]["partial"] is True
+    assert approved["albums"][0]["eligible_count"] == 1
+    # The approval is part of the token, so it can never be replayed as a full run.
+    assert approved["token"] != blocked["token"]
+
+
+def test_partial_approval_cannot_rescue_an_album_with_nothing_ready(tmp_path, immich_server):
+    url, _ = immich_server
+    config, sf, lib = _env(tmp_path, url)
+    _album(sf, lib, decision=None)
+
+    report = _service(sf, config).preflight(allow_partial=True)
+
+    assert report["state"] == "blocked"
+    assert "empty_scope" in {issue["code"] for issue in report["albums"][0]["blockers"]}
+
+
+# ── token ────────────────────────────────────────────────────────────────────
+
+
+def test_token_is_stable_while_nothing_relevant_changes(tmp_path, immich_server):
+    url, _ = immich_server
+    config, sf, lib = _env(tmp_path, url)
+    _album(sf, lib)
+    service = _service(sf, config)
+
+    first = service.preflight()["token"]
+
+    assert service.preflight()["token"] == first
+    assert service.verify_token(first) is True
+
+
+@pytest.mark.parametrize("scope", [None, ["rome"]])
+def test_edited_bytes_make_the_token_stale(tmp_path, immich_server, scope):
+    url, _ = immich_server
+    config, sf, lib = _env(tmp_path, url)
+    folder, _ = _album(sf, lib)
+    service = _service(sf, config)
+    token = service.preflight(scope)["token"]
+
+    (folder / "b.jpg").write_bytes(b"edited outside the app")
+
+    assert service.verify_token(token, scope) is False
+
+
+def test_changed_decision_makes_the_token_stale(tmp_path, immich_server):
+    url, _ = immich_server
+    config, sf, lib = _env(tmp_path, url)
+    _, ids = _album(sf, lib)
+    service = _service(sf, config)
+    token = service.preflight()["token"]
+
+    with sf() as session:
+        session.add(
+            SafetyReview(
+                id=str(uuid.uuid4()),
+                asset_id=ids[0],
+                decision="deferred",
+                created_at=datetime(2099, 1, 1, tzinfo=timezone.utc),  # the latest review wins
+            )
+        )
+        session.commit()
+
+    assert service.verify_token(token) is False
+
+
+def test_token_from_a_different_scope_is_rejected(tmp_path, immich_server):
+    url, _ = immich_server
+    config, sf, lib = _env(tmp_path, url)
+    _album(sf, lib, "rome")
+    _album(sf, lib, "paris", names=("c.jpg",))
+    service = _service(sf, config)
+
+    assert service.verify_token(service.preflight(["rome"])["token"], ["paris"]) is False
+    assert service.verify_token("v1:not-a-real-token") is False
+    assert service.verify_token("") is False
+
+
+# ── API surface ──────────────────────────────────────────────────────────────
+
+
+def test_api_preflight_returns_the_report_without_secrets(tmp_path, immich_server):
+    url, _ = immich_server
+    config, sf, lib = _env(tmp_path, url)
+    _album(sf, lib)
+
+    with TestClient(create_app(config)) as client:
+        response = client.post("/api/v1/upload-preflight", json={})
+
+    assert response.status_code == 200
+    body = response.json()
+    assert body["state"] == "ready" and body["token"].startswith("v1:")
+    assert SENTINEL_KEY not in response.text
+
+
+def test_api_rejects_an_unknown_album(tmp_path, immich_server):
+    url, _ = immich_server
+    config, sf, lib = _env(tmp_path, url)
+    _album(sf, lib)
+
+    with TestClient(create_app(config)) as client:
+        response = client.post("/api/v1/upload-preflight", json={"albums": ["atlantis"]})
+
+    assert response.status_code == 422
+    assert response.json()["error"]["code"] == "unknown_album"

tests/story_traceability.json

@@ -91,6 +91,24 @@
     "US04-02": [
       "tests/unit/test_rename_journal_states.py",
       "tests/integration/test_rename_journal.py"
+    ],
+    "US04-03": [
+      "tests/integration/test_rename_apply.py"
+    ],
+    "US04-04": [
+      "tests/integration/test_rename_recovery.py"
+    ],
+    "US04-05": [
+      "tests/e2e/test_renames_ui.py"
+    ],
+    "US04-06": [
+      "tests/e2e/test_phase_d_pipeline.py"
+    ],
+    "US05-01": [
+      "tests/integration/test_upload_preflight.py"
+    ],
+    "US05-02": [
+      "tests/integration/test_upload_batches.py"
     ]
   }
 }