US04-06: Automate Phase D End-to-End Acceptance

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"

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/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
         ],

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/story_traceability.json

@@ -97,6 +97,12 @@
     ],
     "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"
     ]
   }
 }