US06-06: Automate Phase F End-to-End Acceptance

README.md

@@ -169,3 +169,42 @@ work_item/scripts/python -m pytest -m phase_e -q
   bytes, and recovery after a restart).
 
 Phases A–D remain green in the full run above.
+
+### Phase F acceptance gate
+
+Phase F (Epic E06: archive lifecycle) is the only stage that *removes* originals
+from the library, and the only one whose storage can walk away in someone's bag.
+One command runs the archive fault-injection suites, the black-box archive and
+restore API journeys, and the browser suite:
+
+```bash
+work_item/scripts/python -m pytest -m phase_f -q
+```
+
+- `tests/integration/test_archive_*.py` and `tests/integration/test_restore.py`
+  drive real files on real filesystems: preflight against a mounted, missing,
+  swapped, read-only, or full medium; copy-verify-remove and the same-filesystem
+  move path; and a crash at **every** persisted journal transition in both transfer
+  modes, asserting that no source is ever removed without a durable, byte-identical
+  archive copy.
+- `tests/e2e/test_phase_f_pipeline.py` drives a real server and a real durable
+  worker over HTTP: preflight blockers (offline medium, wrong volume, insufficient
+  capacity, bytes changed after upload), a verified archive whose manifest, hashes,
+  and path history are checked on the medium itself, a worker killed at each of
+  `transferring`, `verified`, `removing`, `source_removed`, and `complete`, the
+  evidence-based recovery that follows, offline deduplication of an exact and a
+  fuzzy copy while the medium is away, mount return, restore, and a collision that
+  restores beside its occupant.
+- **Archived is not missing.** An unmounted medium leaves its photos
+  `archived_offline` — still hashed, still in the duplicate indexes, still
+  previewable through their protected thumbnails — and a rescan neither prunes nor
+  flags them.
+- **Ambiguity is never guessed.** A journal state the medium contradicts stays
+  `manual`, offers no automatic action, and keeps blocking further archiving until
+  a human decides.
+- `tests/e2e/test_archive_ui.py` covers the browser journeys (preview with
+  destination identity and reclaimable bytes, blockers and mount instructions,
+  progress split into transfer/verification/removal, interruption and recovery,
+  offline browsing, restore, collision, keyboard confirmation, and reload).
+
+Phases A–E remain green in the full run above.

frontend/index.html

@@ -19,6 +19,7 @@
         <a href="#/albums" data-nav="albums">Albums</a>
         <a href="#/renames" data-nav="renames">Renames</a>
         <a href="#/uploads" data-nav="uploads">Upload</a>
+        <a href="#/archive" data-nav="archive">Archive</a>
         <a href="#/stats" data-nav="stats">Stats</a>
       </nav>
     </header>

frontend/js/api.js

@@ -138,4 +138,31 @@ export const api = {
     }),
   uploadVerifications: (id, opts = {}) =>
     request(`/upload-batches/${encodeURIComponent(id)}/verifications`, opts),
+
+  // ── Archive and restore: destinations, preflight, plans, recovery ────────
+  archiveLocations: (opts = {}) => request("/archive-locations", opts),
+  registerArchiveLocation: (payload, opts = {}) =>
+    request("/archive-locations", { method: "POST", body: JSON.stringify(payload), ...opts }),
+  archivePreflight: (payload, opts = {}) =>
+    request("/archive-preflight", { method: "POST", body: JSON.stringify(payload), ...opts }),
+  createArchivePlan: (payload, opts = {}) =>
+    request("/archive-plans", { method: "POST", body: JSON.stringify(payload), ...opts }),
+  listArchivePlans: (opts = {}) => request("/archive-plans", opts),
+  getArchivePlan: (id, opts = {}) => request(`/archive-plans/${encodeURIComponent(id)}`, opts),
+  applyArchivePlan: (id, opts = {}) =>
+    request(`/archive-plans/${encodeURIComponent(id)}/apply`, { method: "POST", ...opts }),
+  archiveRecovery: (opts = {}) => request("/archive-recovery", opts),
+  resolveArchiveRecovery: (opts = {}) =>
+    request("/archive-recovery/resolve", { method: "POST", ...opts }),
+  restorePreflight: (payload, opts = {}) =>
+    request("/restore-preflight", { method: "POST", body: JSON.stringify(payload), ...opts }),
+  createRestorePlan: (payload, opts = {}) =>
+    request("/restore-plans", { method: "POST", body: JSON.stringify(payload), ...opts }),
+  listRestorePlans: (opts = {}) => request("/restore-plans", opts),
+  getRestorePlan: (id, opts = {}) => request(`/restore-plans/${encodeURIComponent(id)}`, opts),
+  applyRestorePlan: (id, opts = {}) =>
+    request(`/restore-plans/${encodeURIComponent(id)}/apply`, { method: "POST", ...opts }),
+  restoreRecovery: (opts = {}) => request("/restore-recovery", opts),
+  resolveRestoreRecovery: (opts = {}) =>
+    request("/restore-recovery/resolve", { method: "POST", ...opts }),
 };

frontend/js/app.js

@@ -1,4 +1,5 @@
 import { api } from "./api.js";
+import { renderArchive, setArchiveRender } from "./archive.js";
 import { navigate, onRouteChange, parseHash } from "./router.js";
 import { renderRenames, setRenamesRender } from "./renames.js";
 import { renderUploads, setUploadsRender } from "./uploads.js";
@@ -366,6 +367,7 @@ function render() {
   else if (path === "/albums") renderAlbums(root, params);
   else if (path === "/renames") renderRenames(root, params);
   else if (path === "/uploads") renderUploads(root, params);
+  else if (path === "/archive") renderArchive(root, params);
   else if (path === "/stats") renderStats(root, params);
   else show(errorBanner("Unknown view"));
 }
@@ -374,5 +376,6 @@ function render() {
 setRender(render);
 setRenamesRender(render);
 setUploadsRender(render);
+setArchiveRender(render);
 onRouteChange(render);
 render();

frontend/js/archive.js

@@ -0,0 +1,870 @@
+// Archive view (US06-05): preview what would leave active storage, confirm it
+// exactly, watch the transfer, recover an interrupted one, browse what is already
+// archived, and bring it back.
+//
+// Archiving is the only stage that removes originals, so this view never decides
+// anything itself: the destination's identity, every blocker, the confirmation
+// token, and what recovery may do all come from the server, and an action the
+// server would refuse is not offered. Two things follow from that. A medium that
+// is not mounted produces an instruction naming it rather than a disabled mystery,
+// and an operation whose evidence is ambiguous offers no button at all.
+import { api } from "./api.js";
+import { el, errorBanner, setActiveNav } from "./dom.js";
+import { subscribeJob } from "./events.js";
+import { navigate } from "./router.js";
+
+let outcome = null;
+let activity = [];
+
+let render = () => {};
+export function setArchiveRender(fn) {
+  render = fn;
+}
+
+// The per-file journal states, split into the three things an operator actually
+// wants told apart: bytes moving, bytes proven, original removed (concept §9).
+const PHASES = [
+  ["planned", "planned", "waiting"],
+  ["transferring", "transfer", "copying to the medium"],
+  ["verified", "verified", "archive copy hashed and manifested"],
+  ["removing", "removing", "removing the active original"],
+  ["complete", "complete", "archived and removed"],
+  ["failed", "failed", "left alone for a decision"],
+];
+
+const AVAILABILITY_LABEL = {
+  active: "in the library",
+  archived_online: "archived · medium mounted",
+  archived_offline: "archived · medium away",
+  missing_unexpected: "missing — unexplained",
+};
+
+export async function renderArchive(root, params = {}) {
+  setActiveNav("archive");
+
+  let locations;
+  try {
+    locations = (await api.archiveLocations()).locations;
+  } catch (error) {
+    root.replaceChildren(errorBanner(`Failed to load archive locations: ${error.message}`));
+    return;
+  }
+  const location = locations.find((l) => l.id === params.location) || locations[0] || null;
+
+  const nodes = [el("h1", {}, "Archive"), locationsCard(locations, location, params)];
+  if (!location) {
+    nodes.push(
+      el(
+        "p",
+        { class: "muted", "data-testid": "no-locations" },
+        "Register the disk, NAS share, or removable medium that will hold archived originals."
+      ),
+      outcomeBanner()
+    );
+    root.replaceChildren(...nodes.filter(Boolean));
+    return;
+  }
+
+  const [preflight, archivePlans, restorePlans, recovery, restoreRecovery, archived, restore] =
+    await Promise.all([
+      load(() => api.archivePreflight({ location_id: location.id })),
+      load(() => api.listArchivePlans()),
+      load(() => api.listRestorePlans()),
+      load(() => api.archiveRecovery()),
+      load(() => api.restoreRecovery()),
+      load(() => api.listAssets({ limit: 200 })),
+      load(() => api.restorePreflight({ location_id: location.id })),
+    ]);
+
+  // Both directions share the runs table: one lane moves these originals, so one
+  // history is what an operator has to reason about.
+  const runs = [...plans(archivePlans), ...plans(restorePlans)].sort((a, b) =>
+    (a.created_at || "").localeCompare(b.created_at || "")
+  );
+  const chosen = runs.find((run) => run.id === params.plan) || runs[runs.length - 1] || null;
+  const selectedPlan = chosen ? await load(() => planDetailOf(chosen)) : null;
+
+  nodes.push(
+    preflight ? previewSection(preflight, location) : null,
+    preflight ? confirmBlock(preflight, location) : null,
+    outcomeBanner(),
+    activityLog(),
+    recoverySection(mergeRecovery(recovery, restoreRecovery)),
+    planList(runs, chosen && chosen.id),
+    selectedPlan ? planDetail(selectedPlan) : null,
+    archived ? archivedSection(archived.items, locations) : null,
+    restore ? restoreSection(restore, location) : null
+  );
+  root.replaceChildren(...nodes.filter(Boolean));
+}
+
+function plans(listed) {
+  return listed ? listed.plans : [];
+}
+
+function planDetailOf(run) {
+  return run.direction === "restore" ? api.getRestorePlan(run.id) : api.getArchivePlan(run.id);
+}
+
+// Archive and restore recovery answer the same question about the same lane, so
+// they are one list; an unresolved item of either kind blocks the other.
+function mergeRecovery(archive, restore) {
+  if (!archive && !restore) return null;
+  return {
+    operations: [...((archive || {}).operations || []), ...((restore || {}).operations || [])],
+    manual: [...((archive || {}).manual || []), ...((restore || {}).manual || [])],
+  };
+}
+
+// A section whose data failed to load must not take the rest of the view with it:
+// the medium being away is exactly when the archived-asset list matters most.
+async function load(call) {
+  try {
+    return await call();
+  } catch (_) {
+    return null;
+  }
+}
+
+// ── locations ────────────────────────────────────────────────────────────────
+// A location is a medium, not a path: the marker's ``media_id`` is what proves the
+// right disk is mounted, so it is shown next to the state it produced.
+function locationsCard(locations, selected, params) {
+  const name = el("input", {
+    type: "text",
+    "data-testid": "location-name",
+    "aria-label": "Archive location name",
+    placeholder: "External disk",
+  });
+  const root = el("input", {
+    type: "text",
+    "data-testid": "location-root",
+    "aria-label": "Archive location path",
+    placeholder: "/Volumes/archive",
+  });
+
+  return el(
+    "div",
+    { class: "card", "data-testid": "archive-locations" },
+    el("h2", {}, "Destinations"),
+    locations.length
+      ? el(
+          "table",
+          { class: "grid", "data-testid": "locations" },
+          el(
+            "thead",
+            {},
+            el(
+              "tr",
+              {},
+              ...["", "Name", "Root", "Medium", "State", "Last seen"].map((label) =>
+                el("th", { scope: "col" }, label)
+              )
+            )
+          ),
+          el(
+            "tbody",
+            {},
+            ...locations.map((location) =>
+              el(
+                "tr",
+                {
+                  "data-testid": "location-row",
+                  "data-name": location.name,
+                  "aria-current": selected && location.id === selected.id ? "true" : false,
+                },
+                el(
+                  "td",
+                  {},
+                  el("input", {
+                    type: "radio",
+                    name: "archive-location",
+                    "data-testid": "select-location",
+                    "aria-label": `Use ${location.name}`,
+                    checked: selected && location.id === selected.id ? "checked" : false,
+                    onchange: () => navigate("/archive", { ...params, location: location.id }),
+                  })
+                ),
+                el("td", { "data-testid": "location-label" }, location.name),
+                el("td", { class: "path", "data-testid": "location-path" }, location.root),
+                el("td", { class: "path", "data-testid": "location-media" }, location.media_id),
+                el(
+                  "td",
+                  {},
+                  el(
+                    "span",
+                    {
+                      class: `badge ${location.state === "online" ? "complete" : "attention"}`,
+                      "data-testid": "location-state",
+                    },
+                    location.state
+                  )
+                ),
+                el("td", { class: "muted" }, location.last_seen_at || "never")
+              )
+            )
+          )
+        )
+      : null,
+    selected && selected.state !== "online" ? mountInstruction(selected) : null,
+    el(
+      "div",
+      { class: "toolbar" },
+      name,
+      root,
+      el(
+        "button",
+        {
+          "data-testid": "register-location",
+          onclick: () =>
+            run(() => api.registerArchiveLocation({ name: name.value, root: root.value })),
+        },
+        "Register destination"
+      )
+    )
+  );
+}
+
+// The one thing the app cannot do for the user: name the medium to connect.
+function mountInstruction(location) {
+  const detail =
+    location.state === "wrong_volume"
+      ? `A different medium is mounted at ${location.root}.`
+      : `Nothing is mounted at ${location.root}.`;
+  return el(
+    "div",
+    { class: "confirm", role: "status", "data-testid": "mount-instruction" },
+    `${detail} Connect “${location.name}” (medium ${location.media_id}) and mount it there, ` +
+      "then reload this view. Archived photos stay listed and searchable meanwhile."
+  );
+}
+
+// ── preview ──────────────────────────────────────────────────────────────────
+function previewSection(preflight, location) {
+  const totals = preflight.totals;
+  const capacity = preflight.capacity;
+  const rows = preflight.albums.map((album) =>
+    el(
+      "tr",
+      { "data-testid": "archive-album-row", "data-album": album.album },
+      el("td", { "data-testid": "album-name" }, album.album),
+      el("td", { class: "path", "data-testid": "album-folder" }, album.folder),
+      el("td", { class: "path", "data-testid": "album-destination" }, album.destination),
+      el(
+        "td",
+        { "data-testid": "album-method" },
+        album.transfer_method === "move" ? "move (same filesystem)" : "copy · verify · remove"
+      ),
+      el("td", { "data-testid": "album-assets" }, String(album.asset_count)),
+      el("td", { "data-testid": "album-reclaim" }, bytes(album.reclaimable_bytes)),
+      el(
+        "td",
+        {},
+        el("span", { class: `badge ${album.state}`, "data-testid": "album-state" }, album.state),
+        ...album.blockers.map((blocker) =>
+          el(
+            "div",
+            { class: "blocker", "data-testid": "album-blocker", "data-code": blocker.code },
+            blocker.message
+          )
+        )
+      )
+    )
+  );
+
+  return el(
+    "div",
+    { "data-testid": "archive-preview" },
+    el("h2", {}, "Preview"),
+    el(
+      "div",
+      { class: "decision-bar" },
+      el(
+        "span",
+        { class: "badge", "data-testid": "destination-identity" },
+        `${location.name} · ${location.media_id}`
+      ),
+      el("span", { class: "badge", "data-testid": "total-albums" }, `${totals.albums} album(s)`),
+      el("span", { class: "badge", "data-testid": "total-assets" }, `${totals.assets} photo(s)`),
+      el(
+        "span",
+        { class: "badge", "data-testid": "total-reclaim" },
+        `${bytes(totals.bytes)} reclaimable`
+      ),
+      el(
+        "span",
+        {
+          class: `badge ${capacity.sufficient ? "complete" : "blocked"}`,
+          "data-testid": "capacity",
+        },
+        `free ${bytes(capacity.free_bytes)} · reserve ${bytes(capacity.reserve_bytes)}`
+      )
+    ),
+    blockerList(preflight.blockers, "preflight-blockers", "preflight-blocker", "This scope cannot be archived yet"),
+    rows.length
+      ? el(
+          "table",
+          { class: "grid", "data-testid": "archive-albums" },
+          el(
+            "thead",
+            {},
+            el(
+              "tr",
+              {},
+              ...["Album", "Folder", "Destination", "Transfer", "Photos", "Reclaims", "State"].map(
+                (label) => el("th", { scope: "col" }, label)
+              )
+            )
+          ),
+          el("tbody", {}, ...rows)
+        )
+      : el(
+          "p",
+          { class: "muted", "data-testid": "no-albums" },
+          "No album has a verified upload whose bytes are still unchanged, so nothing may be archived."
+        )
+  );
+}
+
+function blockerList(blockers, containerId, itemId, title) {
+  if (!blockers || !blockers.length) return null;
+  return el(
+    "div",
+    { class: "alert", role: "alert", "data-testid": containerId },
+    el("strong", {}, title),
+    el(
+      "ul",
+      {},
+      ...blockers.map((blocker) =>
+        el("li", { "data-testid": itemId, "data-code": blocker.code }, `${blocker.code}: ${blocker.message}`)
+      )
+    )
+  );
+}
+
+// ── confirmation ─────────────────────────────────────────────────────────────
+function confirmBlock(preflight, location) {
+  const ready = preflight.state === "ready";
+  const totals = preflight.totals;
+  return el(
+    "div",
+    { class: "card", "data-testid": "confirm" },
+    el("h2", {}, "Confirm"),
+    el(
+      "p",
+      { class: "muted", "data-testid": "confirm-token" },
+      `Preflight ${preflight.token.slice(0, 20)}… · ${location.name}`
+    ),
+    el(
+      "p",
+      { "data-testid": "archive-note" },
+      "Archiving removes each original from the library — but only after its copy on " +
+        "the medium has been written, hashed, and recorded in the manifest. The photos " +
+        "stay searchable and deduplicable while the medium is away, and can be restored " +
+        "from this view."
+    ),
+    el(
+      "div",
+      { class: "toolbar" },
+      el(
+        "button",
+        {
+          class: "primary",
+          "data-testid": "start-archive",
+          disabled: ready ? false : "disabled",
+          title: ready ? false : "resolve the blockers above first",
+          onclick: () =>
+            run(async () => {
+              const plan = await api.createArchivePlan({
+                location_id: location.id,
+                token: preflight.token,
+              });
+              const started = await api.applyArchivePlan(plan.id);
+              watch(started.job.id, plan.id, "archive");
+              return { archiving: plan.asset_count };
+            }),
+        },
+        `Archive ${totals.ready_albums} album(s) · reclaim ${bytes(totals.bytes)}`
+      )
+    )
+  );
+}
+
+// ── plans and progress ───────────────────────────────────────────────────────
+function planList(runs, selectedId) {
+  if (!runs.length) {
+    return el("p", { class: "muted", "data-testid": "no-plans" }, "Nothing has been archived yet.");
+  }
+  return el(
+    "div",
+    { "data-testid": "archive-plans" },
+    el("h2", {}, "Runs"),
+    el(
+      "table",
+      { class: "grid", "data-testid": "plans" },
+      el(
+        "thead",
+        {},
+        el(
+          "tr",
+          {},
+          ...["Created", "Direction", "State", "Photos", "Bytes"].map((l) =>
+            el("th", { scope: "col" }, l)
+          )
+        )
+      ),
+      el(
+        "tbody",
+        {},
+        ...runs.map((plan) =>
+          el(
+            "tr",
+            {
+              "data-testid": "plan-row",
+              "data-plan": plan.id,
+              "data-direction": plan.direction,
+              "aria-current": plan.id === selectedId ? "true" : false,
+            },
+            el(
+              "td",
+              {},
+              el("a", { class: "link", href: `#/archive?plan=${encodeURIComponent(plan.id)}` }, plan.created_at || plan.id)
+            ),
+            el("td", { "data-testid": "plan-direction" }, plan.direction),
+            el("td", {}, el("span", { class: `badge ${plan.state}`, "data-testid": "plan-state" }, plan.state)),
+            el("td", {}, String(plan.asset_count)),
+            el("td", {}, bytes(plan.byte_size))
+          )
+        )
+      )
+    )
+  );
+}
+
+function planDetail(plan) {
+  const operations = plan.operations || [];
+  const counts = {};
+  for (const operation of operations) {
+    counts[operation.journal_state] = (counts[operation.journal_state] || 0) + 1;
+  }
+  return el(
+    "div",
+    { class: "card", "data-testid": "plan-detail", "data-plan": plan.id },
+    el("h2", {}, `${plan.direction === "restore" ? "Restore" : "Archive"} run ${plan.created_at || plan.id}`),
+    // Transfer, verification, and removal are separate answers to separate
+    // questions: what has moved, what is proven, and what is already gone.
+    el(
+      "div",
+      { class: "decision-bar", "data-testid": "plan-progress" },
+      el("span", { class: `badge ${plan.state}`, "data-testid": "detail-state" }, plan.state),
+      ...PHASES.map(([key, label, title]) =>
+        el(
+          "span",
+          { class: `badge ${key}`, "data-testid": `count-${label}`, title },
+          `${label}: ${counts[key] ?? 0}`
+        )
+      )
+    ),
+    el(
+      "table",
+      { class: "grid", "data-testid": "operations" },
+      el(
+        "thead",
+        {},
+        el(
+          "tr",
+          {},
+          ...["Photo", "Destination", "Phase", "Attempts", "Problem"].map((l) =>
+            el("th", { scope: "col" }, l)
+          )
+        )
+      ),
+      el(
+        "tbody",
+        {},
+        ...operations.map((operation) =>
+          el(
+            "tr",
+            { "data-testid": "operation-row", "data-asset-id": operation.asset_id },
+            el("td", { class: "path", "data-testid": "operation-source" }, operation.source_path),
+            el("td", { class: "path", "data-testid": "operation-destination" }, operation.destination_path),
+            el(
+              "td",
+              {},
+              el(
+                "span",
+                { class: `badge ${operation.journal_state}`, "data-testid": "operation-phase" },
+                phaseLabel(operation.journal_state)
+              )
+            ),
+            el("td", {}, String(operation.attempt_count)),
+            el(
+              "td",
+              { class: "muted", "data-testid": "operation-error", "data-code": operation.error_code || "" },
+              operation.error_code ? `${operation.error_code}: ${operation.error_message || ""}` : "—"
+            )
+          )
+        )
+      )
+    )
+  );
+}
+
+function phaseLabel(state) {
+  const found = PHASES.find(([key]) => key === state);
+  return found ? found[1] : state;
+}
+
+// ── recovery ─────────────────────────────────────────────────────────────────
+// What an interrupted run left behind, straight from the journal plus the files on
+// disk. Only the operations the server itself classified as resolvable get an
+// action; ambiguous ones are shown with their evidence and no button.
+function recoverySection(recovery) {
+  const operations = recovery ? recovery.operations : [];
+  if (!operations.length) return null;
+  const manual = recovery.manual || [];
+  const resolvable = operations.length - manual.length;
+
+  return el(
+    "div",
+    { class: "card", "data-testid": "archive-recovery" },
+    el("h2", {}, "Interrupted work"),
+    el(
+      "p",
+      { "data-testid": "recovery-summary" },
+      `${operations.length} operation(s) did not finish · ${resolvable} resolvable · ` +
+        `${manual.length} need a decision`
+    ),
+    el(
+      "table",
+      { class: "grid", "data-testid": "recovery-operations" },
+      el(
+        "thead",
+        {},
+        el(
+          "tr",
+          {},
+          ...["Photo", "Phase", "Verdict", "Evidence"].map((l) => el("th", { scope: "col" }, l))
+        )
+      ),
+      el(
+        "tbody",
+        {},
+        ...operations.map((verdict) =>
+          el(
+            "tr",
+            {
+              "data-testid": "recovery-row",
+              "data-classification": verdict.classification,
+              "data-direction": verdict.direction,
+            },
+            el("td", { class: "path", "data-testid": "recovery-source" }, verdict.source_path),
+            el("td", {}, el("span", { class: "badge" }, phaseLabel(verdict.journal_state))),
+            el(
+              "td",
+              {},
+              el(
+                "span",
+                {
+                  class: `badge ${verdict.classification === "manual" ? "blocked" : "attention"}`,
+                  "data-testid": "recovery-verdict",
+                },
+                verdict.classification
+              )
+            ),
+            el(
+              "td",
+              { class: "muted", "data-testid": "recovery-reason" },
+              `${verdict.reason} (source ${verdict.source_exists ? "present" : "absent"}, ` +
+                `archive copy ${verdict.destination_matches ? "verified" : verdict.destination_exists ? "different bytes" : "absent"})`
+            )
+          )
+        )
+      )
+    ),
+    manual.length
+      ? el(
+          "div",
+          { class: "alert", role: "alert", "data-testid": "recovery-manual" },
+          `${manual.length} operation(s) cannot be resolved from the evidence. Nothing will be ` +
+            "removed or retried for them here: inspect the medium and the library, then decide."
+        )
+      : null,
+    el(
+      "div",
+      { class: "toolbar" },
+      resolvable
+        ? el(
+            "button",
+            {
+              class: "primary",
+              "data-testid": "resolve-recovery",
+              onclick: () => run(() => api.resolveArchiveRecovery()),
+            },
+            `Finish ${resolvable} recoverable operation(s)`
+          )
+        : el(
+            "span",
+            { class: "muted", "data-testid": "no-safe-recovery" },
+            "No operation can be finished safely from here."
+          )
+    )
+  );
+}
+
+// ── archived assets ──────────────────────────────────────────────────────────
+// Browsing what is already archived, including while the medium is away: the
+// retained protected preview and the recorded hashes are the evidence, so the row
+// stays complete and honest instead of turning into a missing file.
+function archivedSection(assets, locations) {
+  const archived = assets.filter((asset) => asset.availability_state !== "active");
+  if (!archived.length) return null;
+  const names = Object.fromEntries(locations.map((location) => [location.id, location.name]));
+
+  return el(
+    "div",
+    { "data-testid": "archived-assets" },
+    el("h2", {}, "Archived photos"),
+    el(
+      "table",
+      { class: "grid", "data-testid": "archived" },
+      el(
+        "thead",
+        {},
+        el(
+          "tr",
+          {},
+          ...["Preview", "Archived as", "Medium", "Availability", "Size"].map((l) =>
+            el("th", { scope: "col" }, l)
+          )
+        )
+      ),
+      el(
+        "tbody",
+        {},
+        ...archived.map((asset) =>
+          el(
+            "tr",
+            { "data-testid": "archived-row", "data-asset-id": asset.id },
+            el(
+              "td",
+              {},
+              el("img", {
+                "data-testid": "archived-preview",
+                width: 96,
+                loading: "lazy",
+                src: api.thumbnailUrl(asset.id, 256),
+                alt: `Preview of ${asset.archive_path || asset.id}`,
+              })
+            ),
+            el("td", { class: "path", "data-testid": "archived-path" }, asset.archive_path || "—"),
+            el(
+              "td",
+              { "data-testid": "archived-medium" },
+              names[asset.archive_location_id] || asset.archive_location_id || "—"
+            ),
+            el(
+              "td",
+              {},
+              el(
+                "span",
+                {
+                  class: `badge ${asset.availability_state === "archived_online" ? "complete" : "attention"}`,
+                  "data-testid": "archived-availability",
+                },
+                AVAILABILITY_LABEL[asset.availability_state] || asset.availability_state
+              )
+            ),
+            el("td", {}, bytes(asset.byte_size))
+          )
+        )
+      )
+    )
+  );
+}
+
+// ── restore ──────────────────────────────────────────────────────────────────
+function restoreSection(restore, location) {
+  const ready = restore.state === "ready";
+  const items = restore.items || [];
+  if (!items.length && !restore.blockers.length) return null;
+
+  return el(
+    "div",
+    { class: "card", "data-testid": "restore" },
+    el("h2", {}, "Restore"),
+    el(
+      "p",
+      { "data-testid": "restore-note" },
+      "Restoring copies the archived bytes back into the library and leaves the archive " +
+        "copy where it is. A name that is already taken is never overwritten: the photo " +
+        "comes back beside it under a visibly different name."
+    ),
+    blockerList(restore.blockers, "restore-blockers", "restore-blocker", "This restore cannot run yet"),
+    items.length
+      ? el(
+          "table",
+          { class: "grid", "data-testid": "restore-items" },
+          el(
+            "thead",
+            {},
+            el(
+              "tr",
+              {},
+              ...["Archived as", "Comes back as", "Size", "State"].map((l) =>
+                el("th", { scope: "col" }, l)
+              )
+            )
+          ),
+          el(
+            "tbody",
+            {},
+            ...items.map((item) =>
+              el(
+                "tr",
+                { "data-testid": "restore-row", "data-asset-id": item.asset_id },
+                el("td", { class: "path", "data-testid": "restore-source" }, item.archive_path),
+                el("td", { class: "path", "data-testid": "restore-destination" }, item.destination_path || "—"),
+                el("td", {}, bytes(item.byte_size)),
+                el(
+                  "td",
+                  {},
+                  item.blockers.length
+                    ? el(
+                        "span",
+                        {
+                          class: "badge blocked",
+                          "data-testid": "restore-item-blocker",
+                          "data-code": item.blockers[0].code,
+                        },
+                        item.blockers[0].code
+                      )
+                    : el("span", { class: "badge ready", "data-testid": "restore-item-state" }, "ready")
+                )
+              )
+            )
+          )
+        )
+      : null,
+    el(
+      "div",
+      { class: "toolbar" },
+      el(
+        "button",
+        {
+          class: "primary",
+          "data-testid": "start-restore",
+          disabled: ready ? false : "disabled",
+          title: ready ? false : "the medium and every archived copy must check out first",
+          onclick: () =>
+            run(async () => {
+              const plan = await api.createRestorePlan({
+                location_id: location.id,
+                token: restore.token,
+              });
+              const started = await api.applyRestorePlan(plan.id);
+              watch(started.job.id, plan.id, "restore");
+              return { restoring: plan.asset_count };
+            }),
+        },
+        `Restore ${items.length} photo(s) from ${location.name}`
+      )
+    )
+  );
+}
+
+// ── running commands ─────────────────────────────────────────────────────────
+async function run(action) {
+  try {
+    outcome = { kind: "ok", result: await action() };
+  } catch (error) {
+    outcome = error.status === 409 ? { kind: "conflict", error } : { kind: "error", error };
+  }
+  render();
+}
+
+// Live job activity. The plan panel is refreshed on its own tick because the
+// journal advances per file, not per job event; a full re-render would re-run
+// preflight (which re-hashes the library), so that happens once when the job ends.
+const REFRESH_MS = 500;
+
+function watch(jobId, planId, kind) {
+  activity = [`Started ${kind} job ${jobId}`];
+  const tick = setInterval(() => refreshPlan(planId, kind), REFRESH_MS);
+  subscribeJob(jobId, {
+    onEvent: (event) => {
+      activity.push(`${event.type}${event.message ? ": " + event.message : ""}`);
+      const log = document.querySelector('[data-testid="archive-activity"]');
+      if (log) log.textContent = activity.join("\n");
+    },
+    onDone: () => {
+      clearInterval(tick);
+      activity.push("done");
+      render();
+    },
+  });
+}
+
+async function refreshPlan(planId, kind) {
+  const node = document.querySelector(`[data-testid="plan-detail"][data-plan="${planId}"]`);
+  if (!node) return; // the user navigated away from the running plan
+  try {
+    const plan = await (kind === "restore" ? api.getRestorePlan(planId) : api.getArchivePlan(planId));
+    node.replaceWith(planDetail(plan));
+  } catch (_) {
+    // Transient; the next tick tries again and the job's end re-renders anyway.
+  }
+}
+
+function activityLog() {
+  return el(
+    "pre",
+    {
+      class: "activity-log",
+      role: "status",
+      "aria-live": "polite",
+      "data-testid": "archive-activity",
+    },
+    activity.join("\n")
+  );
+}
+
+function outcomeBanner() {
+  if (!outcome) return null;
+  if (outcome.kind === "conflict") {
+    return el(
+      "div",
+      { class: "alert", role: "alert", "data-testid": "conflict" },
+      `The server refused this: ${outcome.error.message}. Nothing was moved or removed; ` +
+        "the state below is the server's current one — review it and decide again."
+    );
+  }
+  if (outcome.kind === "error") {
+    return el(
+      "div",
+      { class: "alert", role: "alert", "data-testid": "archive-error" },
+      `Failed: ${outcome.error.message}`
+    );
+  }
+  const result = outcome.result || {};
+  const message =
+    result.archiving !== undefined
+      ? `Archiving ${result.archiving} photo(s). Originals are removed only after their copies verify.`
+      : result.restoring !== undefined
+        ? `Restoring ${result.restoring} photo(s) into the library.`
+        : "Done — the state below is the server's.";
+  return el("div", { class: "alert", role: "status", "data-testid": "archive-result" }, message);
+}
+
+// ── formatting ───────────────────────────────────────────────────────────────
+function bytes(value) {
+  if (value == null) return "unknown";
+  const units = ["B", "kB", "MB", "GB", "TB"];
+  let size = value;
+  let unit = 0;
+  while (size >= 1000 && unit < units.length - 1) {
+    size /= 1000;
+    unit += 1;
+  }
+  return `${unit === 0 ? size : size.toFixed(1)} ${units[unit]}`;
+}

photo_pipeline/services/archives.py

@@ -208,9 +208,12 @@ class ArchiveService:
             report["blockers"].append(
                 _issue(
                     "insufficient_capacity",
+                    # The free-space number is deliberately left out: it drifts between
+                    # two identical preflights, and the token is a digest of this text,
+                    # so quoting it here would invalidate every approval instantly.
                     f"{report['totals']['bytes']} B plus a "
-                    f"{self._config.archive_free_space_reserve_bytes} B reserve do not fit in "
+                    f"{self._config.archive_free_space_reserve_bytes} B reserve do not fit on "
-                    f"{report['capacity']['free_bytes']} B of free space",
+                    "the medium",
                 )
             )
         report["backup"] = self._backup_probe()

photo_pipeline/services/restores.py

@@ -140,9 +140,11 @@ class RestoreService:
             report["blockers"].append(
                 _issue(
                     "insufficient_capacity",
+                    # No free-space number here: it drifts between two identical
+                    # preflights and the token is a digest of this text (US06-06).
                     f"{report['totals']['bytes']} B plus a "
                     f"{self._config.archive_free_space_reserve_bytes} B reserve do not fit in "
-                    f"{report['capacity']['free_bytes']} B of free space",
+                    "the library",
                 )
             )
         if not items:

tests/e2e/_pipeline_harness.py

@@ -393,6 +393,125 @@ def mark_upload_ready(seeded: Seeded, *, unverified: tuple[str, ...] = ()) -> No
             session.commit()
 
 
+# ── Phase F: an archivable album and a mountable fake medium ─────────────────
+
+
+def mark_uploaded(seeded: Seeded, *, album: str = "rome") -> None:
+    """Give every seeded photo the verified upload evidence archiving requires.
+
+    Archiving refuses anything Immich is not proven to hold, and that proof is an
+    upload batch — recorded here as fixture state so the archive journeys do not
+    have to re-run an upload they are not testing.
+    """
+    from sqlalchemy import select
+
+    from photo_pipeline.models import Asset, UploadBatch, UploadItem
+    from photo_pipeline.services.hashing import sha256_file
+
+    with session_factory(seeded) as sf:
+        with sf() as session:
+            batch_id = str(uuid.uuid4())
+            session.add(
+                UploadBatch(
+                    id=batch_id,
+                    album=album,
+                    folder=str(seeded.lib / album),
+                    album_name=album,
+                    state="succeeded",
+                    preflight_token="v1:e2e",
+                    outcome_state="verified",
+                    created_at=NOW,
+                )
+            )
+            for asset in session.scalars(select(Asset)):
+                if not asset.current_path:
+                    continue
+                session.add(
+                    UploadItem(
+                        batch_id=batch_id,
+                        asset_id=asset.id,
+                        path=asset.current_path,
+                        sha256=sha256_file(asset.current_path),
+                        sha1="0" * 40,
+                        state="sent",
+                        outcome="uploaded",
+                    )
+                )
+            session.commit()
+
+
+class ArchiveStack:
+    """A seeded, archivable library plus the server, the worker, and a fake medium.
+
+    The medium is an ordinary directory whose marker file makes it identifiable;
+    ``unmount()`` takes that marker away, which is exactly what the application sees
+    when an external disk is unplugged.
+    """
+
+    MARKER = ".photo-pipeline-archive.json"
+
+    def __init__(self, tmp_path: Path, seeded: Seeded) -> None:
+        self.tmp_path = tmp_path
+        self.seeded = seeded
+        self.archive = tmp_path / "archive"
+        self.archive.mkdir(exist_ok=True)
+        self.server: Server | None = None
+        self.worker: subprocess.Popen | None = None
+        self.base = ""
+
+    def start(self, *, worker: bool = True, extra_env: dict[str, str] | None = None) -> "ArchiveStack":
+        env = {"PHOTO_PIPELINE_ARCHIVE_FREE_SPACE_RESERVE_BYTES": "0", **(extra_env or {})}
+        self.server = Server(self.seeded, extra_env=env).start()
+        self.base = self.server.base
+        if worker:
+            self.worker = start_worker(self.seeded, extra_env=env)
+        return self
+
+    def register(self, name: str = "external") -> dict:
+        response = httpx.post(
+            f"{self.base}/api/v1/archive-locations",
+            json={"name": name, "root": str(self.archive)},
+            timeout=20,
+        )
+        response.raise_for_status()
+        return response.json()
+
+    def unmount(self) -> None:
+        (self.archive / self.MARKER).rename(self.archive / f"{self.MARKER}.away")
+
+    def remount(self) -> None:
+        (self.archive / f"{self.MARKER}.away").rename(self.archive / self.MARKER)
+
+    def plans(self) -> list[dict]:
+        return httpx.get(f"{self.base}/api/v1/archive-plans", timeout=20).json()["plans"]
+
+    def assets(self) -> list[dict]:
+        return httpx.get(
+            f"{self.base}/api/v1/inventory/assets", params={"limit": 200}, timeout=20
+        ).json()["items"]
+
+    def restart_server(self) -> None:
+        self.server.stop()
+        self.server.start()
+
+    def restart_worker(self, *, extra_env: dict[str, str] | None = None) -> None:
+        """Replace the worker — a healthy one after a crashed one, by default."""
+        if self.worker is not None and self.worker.poll() is None:
+            self.worker.kill()
+            self.worker.wait(timeout=10)
+        self.worker = start_worker(
+            self.seeded,
+            extra_env={"PHOTO_PIPELINE_ARCHIVE_FREE_SPACE_RESERVE_BYTES": "0", **(extra_env or {})},
+        )
+
+    def stop(self) -> None:
+        if self.worker is not None:
+            self.worker.kill()
+            self.worker.wait(timeout=10)
+        if self.server is not None:
+            self.server.stop()
+
+
 class UploadStack:
     """A seeded, upload-ready library plus the server, worker, and fake Immich."""
 

tests/e2e/test_archive_ui.py

@@ -0,0 +1,302 @@
+"""Browser journeys for the archive view (US06-05).
+
+Archiving is the only stage that removes originals, so these journeys check the
+two things a browser must never get wrong about it: that the preview names exactly
+what would leave and where it would go, and that nothing offers an action the
+server would refuse. The medium is a real directory whose marker makes it
+identifiable; unmounting it is what an unplugged disk looks like from here.
+
+Nothing is mocked inside the browser: the transfer runs in the real worker process
+and the assertions read the filesystem afterwards.
+"""
+
+from __future__ import annotations
+
+import pytest
+from playwright.sync_api import expect
+
+from tests.e2e._pipeline_harness import (
+    ArchiveStack,
+    mark_uploaded,
+    seed_album,
+    session_factory,
+    wait_until,
+)
+
+pytestmark = pytest.mark.phase_f  # part of the Phase F acceptance gate (US06-06)
+
+TIMEOUT = 10
+RUN_TIMEOUT = 30_000
+
+
+@pytest.fixture
+def stack(tmp_path):
+    seeded = seed_album(tmp_path)
+    mark_uploaded(seeded)
+    running = ArchiveStack(tmp_path, seeded)
+    try:
+        yield running
+    finally:
+        running.stop()
+
+
+def _open(page, stack) -> None:
+    page.goto(f"{stack.base}/app/#/archive")
+    page.get_by_test_id("archive-locations").wait_for()
+
+
+def _archive(page, stack) -> None:
+    """Confirm the archive and wait for the worker to finish the run."""
+    _open(page, stack)
+    page.get_by_test_id("start-archive").click()
+    expect(page.get_by_test_id("detail-state")).to_have_text("complete", timeout=RUN_TIMEOUT)
+
+
+def _archived_paths(stack) -> list[str]:
+    return sorted(p.name for p in (stack.archive / "rome").glob("*.jpg"))
+
+
+# ── preview and confirmation ─────────────────────────────────────────────────
+
+
+def test_the_preview_names_the_scope_destination_and_reclaimable_bytes(page, stack):
+    errors = []
+    page.on("console", lambda m: errors.append(m.text) if m.type == "error" else None)
+    stack.start(worker=False)
+    location = stack.register()
+    _open(page, stack)
+
+    row = page.get_by_test_id("location-row").first
+    expect(row.get_by_test_id("location-label")).to_have_text("external")
+    expect(row.get_by_test_id("location-media")).to_have_text(location["media_id"])
+    expect(row.get_by_test_id("location-state")).to_have_text("online")
+
+    album = page.get_by_test_id("archive-album-row").first
+    expect(album.get_by_test_id("album-name")).to_have_text("rome")
+    expect(album.get_by_test_id("album-destination")).to_have_text(str(stack.archive / "rome"))
+    expect(album.get_by_test_id("album-assets")).to_have_text("2")
+    expect(album.get_by_test_id("album-state")).to_have_text("ready")
+    # Same filesystem here, so the transfer method is the move path — and it is
+    # named, because copy-verify-remove and move fail differently.
+    expect(album.get_by_test_id("album-method")).to_contain_text("move")
+
+    expect(page.get_by_test_id("destination-identity")).to_contain_text(location["media_id"])
+    expect(page.get_by_test_id("capacity")).to_contain_text("reserve")
+    expect(page.get_by_test_id("start-archive")).to_contain_text("Archive 1 album(s) · reclaim")
+    expect(page.get_by_test_id("start-archive")).to_be_enabled()
+    assert stack.plans() == [], "previewing may not create anything"
+    assert errors == [], f"console errors: {errors}"
+
+
+def test_an_offline_medium_blocks_the_confirmation_and_says_what_to_mount(page, stack):
+    stack.start(worker=False)
+    location = stack.register()
+    stack.unmount()
+    _open(page, stack)
+
+    expect(page.get_by_test_id("location-state")).to_have_text("offline")
+    expect(page.get_by_test_id("preflight-blocker").first).to_have_attribute(
+        "data-code", "location_offline"
+    )
+    instruction = page.get_by_test_id("mount-instruction")
+    expect(instruction).to_contain_text("external")
+    expect(instruction).to_contain_text(location["media_id"])
+    expect(page.get_by_test_id("start-archive")).to_be_disabled()
+
+
+def test_an_unuploaded_album_is_blocked_with_its_reason(page, stack):
+    # No upload evidence at all: archiving would remove the only copy.
+    seeded = stack.seeded
+    with session_factory(seeded) as sf:
+        from sqlalchemy import delete
+
+        from photo_pipeline.models import UploadItem
+
+        with sf() as session:
+            session.execute(delete(UploadItem))
+            session.commit()
+    stack.start(worker=False)
+    stack.register()
+    _open(page, stack)
+
+    expect(page.get_by_test_id("album-state")).to_have_text("blocked")
+    expect(page.get_by_test_id("album-blocker").first).to_have_attribute(
+        "data-code", "partial_scope"
+    )
+    expect(page.get_by_test_id("start-archive")).to_be_disabled()
+
+
+# ── running, progress, reload ────────────────────────────────────────────────
+
+
+def test_a_confirmed_archive_runs_and_separates_transfer_verify_and_removal(page, stack):
+    stack.start()
+    stack.register()
+    _archive(page, stack)
+
+    expect(page.get_by_test_id("count-complete")).to_have_text("complete: 2")
+    expect(page.get_by_test_id("count-failed")).to_have_text("failed: 0")
+    expect(page.get_by_test_id("count-transfer")).to_have_text("transfer: 0")
+    expect(page.get_by_test_id("count-verified")).to_have_text("verified: 0")
+    expect(page.get_by_test_id("count-removing")).to_have_text("removing: 0")
+    expect(page.get_by_test_id("operation-phase").first).to_have_text("complete")
+
+    # The library really lost the originals and the medium really holds them.
+    assert _archived_paths(stack) == ["a.jpg", "b.jpg"]
+    assert not (stack.seeded.lib / "rome" / "a.jpg").exists()
+    assert {a["availability_state"] for a in stack.assets()} == {"archived_online"}
+
+
+def test_the_run_survives_a_reload_because_the_state_is_the_servers(page, stack):
+    stack.start()
+    stack.register()
+    _archive(page, stack)
+
+    page.reload()
+    page.get_by_test_id("plan-detail").wait_for()
+    expect(page.get_by_test_id("detail-state")).to_have_text("complete")
+    expect(page.get_by_test_id("count-complete")).to_have_text("complete: 2")
+    # And the run stays addressable by its own URL.
+    plan_id = stack.plans()[0]["id"]
+    page.goto(f"{stack.base}/app/#/archive?plan={plan_id}")
+    expect(page.get_by_test_id("plan-detail")).to_have_attribute("data-plan", plan_id)
+
+
+# ── recovery ─────────────────────────────────────────────────────────────────
+
+
+def test_an_interrupted_transfer_is_shown_with_evidence_and_a_safe_action(page, stack):
+    stack.start(worker=False)
+    stack.register()
+    _open(page, stack)
+    page.get_by_test_id("start-archive").click()
+    expect(page.get_by_test_id("archive-result")).to_be_visible()
+    # No worker ran, so every item is still planned; leave one mid-transfer as a
+    # crash would: intent recorded, nothing published, source intact.
+    _interrupt(stack)
+
+    page.reload()
+    page.get_by_test_id("archive-recovery").wait_for()
+    row = page.get_by_test_id("recovery-row").first
+    expect(row).to_have_attribute("data-classification", "resumable")
+    expect(row.get_by_test_id("recovery-reason")).to_contain_text("source present")
+    page.get_by_test_id("resolve-recovery").click()
+    expect(page.get_by_test_id("archive-recovery")).to_have_count(0)
+    assert (stack.seeded.lib / "rome" / "a.jpg").exists(), "recovery may not move anything"
+
+
+def test_ambiguous_evidence_offers_no_action_at_all(page, stack):
+    stack.start(worker=False)
+    stack.register()
+    _open(page, stack)
+    page.get_by_test_id("start-archive").click()
+    expect(page.get_by_test_id("archive-result")).to_be_visible()
+    # Journal says the copy is verified, but the medium holds nothing: no evidence
+    # supports either finishing or retrying this item.
+    _interrupt(stack, state="verified")
+
+    page.reload()
+    page.get_by_test_id("archive-recovery").wait_for()
+    expect(page.get_by_test_id("recovery-row").first).to_have_attribute(
+        "data-classification", "manual"
+    )
+    expect(page.get_by_test_id("recovery-manual")).to_be_visible()
+    expect(page.get_by_test_id("resolve-recovery")).to_have_count(0)
+    expect(page.get_by_test_id("no-safe-recovery")).to_be_visible()
+
+
+def _interrupt(stack, *, state: str = "transferring") -> None:
+    """Leave the plan's first operation in a non-terminal journal state."""
+    with session_factory(stack.seeded) as sf:
+        from photo_pipeline.services.archive_journal import ArchiveJournal
+
+        journal = ArchiveJournal(sf)
+        plan_id = stack.plans()[0]["id"]
+        operation = journal.operations(plan_id)[0]
+        journal.begin(operation["id"], worker_id="crashed", fencing_token=1)
+        if state != "transferring":
+            journal.transition(operation["id"], state, fencing_token=1)
+
+
+# ── offline browsing and restore ─────────────────────────────────────────────
+
+
+def test_archived_photos_stay_browsable_while_the_medium_is_away(page, stack):
+    stack.start()
+    stack.register()
+    _archive(page, stack)
+    stack.unmount()
+
+    page.reload()
+    page.get_by_test_id("archived-assets").wait_for()
+    row = page.get_by_test_id("archived-row").first
+    expect(row.get_by_test_id("archived-availability")).to_have_text("archived · medium away")
+    expect(row.get_by_test_id("archived-path")).to_contain_text("rome/")
+    expect(row.get_by_test_id("archived-medium")).to_have_text("external")
+    expect(page.get_by_test_id("mount-instruction")).to_contain_text("external")
+    # The retained protected preview is served even though the original is gone.
+    preview = row.get_by_test_id("archived-preview")
+    assert preview.evaluate("img => img.complete && img.naturalWidth > 0")
+    # Restoring is impossible right now, and says so instead of failing later.
+    expect(page.get_by_test_id("start-restore")).to_be_disabled()
+
+
+def test_restoring_brings_the_photos_back_without_losing_identity(page, stack):
+    stack.start()
+    stack.register()
+    _archive(page, stack)
+    before = {asset["id"] for asset in stack.assets()}
+
+    page.reload()
+    page.get_by_test_id("restore").wait_for()
+    expect(page.get_by_test_id("restore-row").first.get_by_test_id("restore-destination")).to_contain_text(
+        str(stack.seeded.lib / "rome")
+    )
+    page.get_by_test_id("start-restore").click()
+    wait_until(
+        lambda: all(a["availability_state"] == "active" for a in stack.assets()),
+        timeout=30,
+    )
+
+    assert {asset["id"] for asset in stack.assets()} == before  # same identities
+    assert (stack.seeded.lib / "rome" / "a.jpg").exists()
+    assert _archived_paths(stack) == ["a.jpg", "b.jpg"]  # the archive copy stays
+
+
+def test_a_taken_name_is_restored_beside_it_never_over_it(page, stack):
+    stack.start()
+    stack.register()
+    _archive(page, stack)
+    squatter = stack.seeded.lib / "rome" / "a.jpg"
+    squatter.parent.mkdir(parents=True, exist_ok=True)
+    squatter.write_bytes(b"a different photo now lives here")
+
+    page.reload()
+    page.get_by_test_id("restore").wait_for()
+    destinations = page.get_by_test_id("restore-destination").all_inner_texts()
+    assert any("(restored)" in text for text in destinations), destinations
+
+    page.get_by_test_id("start-restore").click()
+    wait_until(
+        lambda: all(a["availability_state"] == "active" for a in stack.assets()),
+        timeout=30,
+    )
+    assert squatter.read_bytes() == b"a different photo now lives here"
+    assert (stack.seeded.lib / "rome" / "a (restored).jpg").exists()
+
+
+# ── keyboard ─────────────────────────────────────────────────────────────────
+
+
+def test_the_whole_archive_can_be_confirmed_from_the_keyboard(page, stack):
+    stack.start()
+    stack.register()
+    _open(page, stack)
+
+    button = page.get_by_test_id("start-archive")
+    button.focus()
+    expect(button).to_be_focused()
+    page.keyboard.press("Enter")
+
+    expect(page.get_by_test_id("detail-state")).to_have_text("complete", timeout=RUN_TIMEOUT)
+    assert _archived_paths(stack) == ["a.jpg", "b.jpg"]

tests/e2e/test_phase_f_pipeline.py

@@ -0,0 +1,505 @@
+"""Phase F end-to-end acceptance (US06-06): archive lifecycle, black box.
+
+Every journey drives a real ``photo_pipeline serve`` child process and a real
+durable worker over HTTP — register a medium, preflight it, archive an album,
+crash mid-transfer, recover, unmount the medium, rediscover the photos through
+their hashes while they are unreachable, mount it again, restore, and collide.
+Nothing is reached into: the medium is an ordinary directory whose marker file is
+its identity, and unmounting it means taking that marker away, which is what the
+application sees when a disk is unplugged.
+
+Three invariants are asserted wherever they apply, because they are what make the
+only *removing* stage safe:
+
+- **No source is removed before its archive copy is verified.** Every crash barrier
+  is checked for the file still being somewhere: either in the library, or on the
+  medium hashing exactly as recorded.
+- **Archived is not missing.** An unmounted medium leaves its photos
+  ``archived_offline``, still hashed, still deduplicable, still previewable.
+- **Restoring never overwrites.** A taken name comes back beside its occupant, and
+  the archived copy stays on the medium.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import shutil
+from pathlib import Path
+
+import httpx
+import pytest
+
+from tests.e2e._pipeline_harness import (
+    ArchiveStack,
+    mark_uploaded,
+    seed_album,
+    session_factory,
+    wait_until,
+)
+
+pytestmark = pytest.mark.phase_f
+
+TIMEOUT = 20
+ALBUM = "rome"
+MANIFEST = "archive-manifest.jsonl"
+# Every persisted transition the transfer can die at, in the order it reaches them.
+BARRIERS = ["transferring", "verified", "removing", "source_removed", "complete"]
+
+
+@pytest.fixture
+def stack(tmp_path):
+    """An analysed album with verified upload evidence — archivable, nothing else."""
+    seeded = seed_album(tmp_path)
+    mark_uploaded(seeded)
+    running = ArchiveStack(tmp_path, seeded)
+    try:
+        yield running
+    finally:
+        running.stop()
+
+
+# ── helpers ──────────────────────────────────────────────────────────────────
+
+
+def _preflight(stack, location_id: str, **body) -> dict:
+    response = httpx.post(
+        f"{stack.base}/api/v1/archive-preflight",
+        json={"location_id": location_id, **body},
+        timeout=TIMEOUT,
+    )
+    response.raise_for_status()
+    return response.json()
+
+
+def _plan(stack, location_id: str, **body) -> dict:
+    report = _preflight(stack, location_id, **body)
+    response = httpx.post(
+        f"{stack.base}/api/v1/archive-plans",
+        json={"location_id": location_id, "token": report["token"], **body},
+        timeout=TIMEOUT,
+    )
+    response.raise_for_status()
+    return response.json()
+
+
+def _apply(stack, plan_id: str) -> httpx.Response:
+    return httpx.post(f"{stack.base}/api/v1/archive-plans/{plan_id}/apply", timeout=TIMEOUT)
+
+
+def _plan_state(stack, plan_id: str) -> dict:
+    return httpx.get(f"{stack.base}/api/v1/archive-plans/{plan_id}", timeout=TIMEOUT).json()
+
+
+def _await_plan(stack, plan_id: str, states=("complete", "failed"), *, timeout: float = 60) -> dict:
+    return wait_until(
+        lambda: (lambda p: p if p.get("state") in states else None)(_plan_state(stack, plan_id)),
+        timeout=timeout,
+    )
+
+
+def _archive_album(stack) -> dict:
+    """Register, preflight, plan, apply, and wait for the run to finish."""
+    location = stack.register()
+    plan = _plan(stack, location["id"])
+    _apply(stack, plan["id"]).raise_for_status()
+    return {"location": location, "plan": _await_plan(stack, plan["id"])}
+
+
+def _assets(stack) -> dict[str, dict]:
+    return {asset["id"]: asset for asset in stack.assets()}
+
+
+def _recovery(stack) -> dict:
+    return httpx.get(f"{stack.base}/api/v1/archive-recovery", timeout=TIMEOUT).json()
+
+
+def _resolve(stack) -> dict:
+    response = httpx.post(f"{stack.base}/api/v1/archive-recovery/resolve", timeout=TIMEOUT)
+    response.raise_for_status()
+    return response.json()
+
+
+def _restore(stack, location_id: str) -> dict:
+    report = httpx.post(
+        f"{stack.base}/api/v1/restore-preflight",
+        json={"location_id": location_id},
+        timeout=TIMEOUT,
+    ).json()
+    plan = httpx.post(
+        f"{stack.base}/api/v1/restore-plans",
+        json={"location_id": location_id, "token": report["token"]},
+        timeout=TIMEOUT,
+    )
+    plan.raise_for_status()
+    plan = plan.json()
+    httpx.post(
+        f"{stack.base}/api/v1/restore-plans/{plan['id']}/apply", timeout=TIMEOUT
+    ).raise_for_status()
+    wait_until(
+        lambda: all(a["availability_state"] == "active" for a in stack.assets()), timeout=60
+    )
+    return {"preflight": report, "plan": plan}
+
+
+def _sha256(path: Path) -> str:
+    return hashlib.sha256(path.read_bytes()).hexdigest()
+
+
+def _contents(*roots: Path) -> set[str]:
+    """Every photo byte-string reachable anywhere, keyed by hash."""
+    return {
+        _sha256(path)
+        for root in roots
+        for path in root.rglob("*.jpg")
+        if path.is_file() and not path.name.startswith(".")
+    }
+
+
+def _manifest(stack) -> list[dict]:
+    path = stack.archive / ALBUM / MANIFEST
+    if not path.exists():
+        return []
+    return [json.loads(line) for line in path.read_text().splitlines() if line.strip()]
+
+
+# ── preflight blockers ───────────────────────────────────────────────────────
+
+
+def test_preflight_refuses_an_offline_medium_a_wrong_volume_and_a_full_disk(stack):
+    stack.start(worker=False)
+    location = stack.register()
+
+    ready = _preflight(stack, location["id"])
+    assert ready["state"] == "ready"
+    assert ready["totals"]["assets"] == 2
+
+    stack.unmount()
+    offline = _preflight(stack, location["id"])
+    assert offline["state"] == "blocked"
+    assert "location_offline" in {issue["code"] for issue in offline["blockers"]}
+
+    # A different disk mounted at the same place is not this medium.
+    (stack.archive / ArchiveStack.MARKER).write_text('{"media_id": "another-disk"}')
+    wrong = _preflight(stack, location["id"])
+    assert "wrong_volume" in {issue["code"] for issue in wrong["blockers"]}
+
+    # Nothing has been created, moved, or removed by any of that.
+    assert stack.plans() == []
+    assert sorted(p.name for p in (stack.seeded.lib / ALBUM).glob("*.jpg")) == ["a.jpg", "b.jpg"]
+
+
+def test_preflight_refuses_capacity_it_cannot_prove_and_bytes_it_cannot_vouch_for(tmp_path):
+    seeded = seed_album(tmp_path)
+    mark_uploaded(seeded)
+    stack = ArchiveStack(tmp_path, seeded)
+    try:
+        # A reserve larger than any disk: the scope cannot be promised room.
+        stack.start(
+            worker=False,
+            extra_env={"PHOTO_PIPELINE_ARCHIVE_FREE_SPACE_RESERVE_BYTES": str(1 << 62)},
+        )
+        location = stack.register()
+        report = _preflight(stack, location["id"])
+        assert "insufficient_capacity" in {issue["code"] for issue in report["blockers"]}
+
+        # And a plan for a blocked scope is refused rather than half-created.
+        refused = httpx.post(
+            f"{stack.base}/api/v1/archive-plans",
+            json={"location_id": location["id"], "token": report["token"]},
+            timeout=TIMEOUT,
+        )
+        assert refused.status_code == 422
+        assert refused.json()["error"]["code"] == "blocked"
+        assert stack.plans() == []
+    finally:
+        stack.stop()
+
+
+def test_an_album_whose_bytes_changed_after_upload_cannot_be_archived(stack):
+    stack.start(worker=False)
+    location = stack.register()
+    (stack.seeded.lib / ALBUM / "a.jpg").write_bytes(b"edited after the upload")
+
+    report = _preflight(stack, location["id"])
+    album = report["albums"][0]
+    assert album["state"] == "blocked"
+    assert "partial_scope" in {issue["code"] for issue in album["blockers"]}
+    assert "bytes_changed" in {
+        issue["code"] for asset in album["assets"] for issue in asset["blockers"]
+    }
+    assert report["state"] == "blocked"
+
+
+# ── copy, verify, remove ─────────────────────────────────────────────────────
+
+
+def test_an_archived_album_leaves_the_library_only_after_it_is_verified(stack):
+    stack.start()
+    before = _contents(stack.seeded.lib)
+    hashes = {a["id"]: a["current_sha256"] for a in stack.assets()}
+
+    result = _archive_album(stack)
+    plan = result["plan"]
+
+    assert plan["state"] == "complete"
+    assert {op["journal_state"] for op in plan["operations"]} == {"complete"}
+    # Same filesystem here, so the transfer took the atomic-move path — recorded as
+    # what actually happened, not as the plan's guess.
+    assert all(op["same_filesystem"] for op in plan["operations"])
+
+    # The bytes are on the medium, hashing exactly as recorded, and gone from the
+    # library. Not one photo was lost in between.
+    assert _contents(stack.archive) == before
+    assert not list((stack.seeded.lib / ALBUM).glob("*.jpg"))
+    for asset_id, sha256 in hashes.items():
+        asset = _assets(stack)[asset_id]
+        assert asset["current_path"] is None
+        assert asset["availability_state"] == "archived_online"
+        assert _sha256(stack.archive / asset["archive_path"]) == sha256
+
+    # The medium carries its own record of what it holds.
+    manifest = _manifest(stack)
+    assert {entry["asset_id"] for entry in manifest} == set(hashes)
+    assert {entry["sha256"] for entry in manifest} == set(hashes.values())
+    assert {entry["media_id"] for entry in manifest} == {result["location"]["media_id"]}
+
+
+def test_the_archived_state_survives_a_server_restart(stack):
+    stack.start()
+    result = _archive_album(stack)
+    before = _assets(stack)
+    plan_before = _plan_state(stack, result["plan"]["id"])
+
+    stack.restart_server()
+
+    assert _assets(stack) == before
+    assert _plan_state(stack, result["plan"]["id"]) == plan_before
+    assert {a["availability_state"] for a in stack.assets()} == {"archived_online"}
+
+
+# ── interruption at every persisted transition ───────────────────────────────
+
+
+@pytest.mark.parametrize("barrier", BARRIERS)
+def test_a_worker_killed_at_each_transition_loses_nothing_and_recovers(tmp_path, barrier):
+    # One photo, so the crash lands on the only operation and the end state is the
+    # recovery's own doing rather than a mixture with untouched work.
+    seeded = seed_album(tmp_path, names=("a.jpg",))
+    mark_uploaded(seeded)
+    stack = ArchiveStack(tmp_path, seeded)
+    try:
+        # Only the worker carries the crash barrier: it dies the instant this journal
+        # state is persisted, while the server stays up to be asked what happened.
+        stack.start(worker=False)
+        stack.restart_worker(extra_env={"PHOTO_PIPELINE_FAULT_AFTER": barrier})
+        before = _contents(stack.seeded.lib)
+        location = stack.register()
+        plan = _plan(stack, location["id"])
+        _apply(stack, plan["id"]).raise_for_status()
+
+        wait_until(lambda: stack.worker.poll() is not None, timeout=60)
+        assert stack.worker.returncode in (9, -9), "the worker should have been killed"
+
+        # Whatever the crash interrupted, every photo is still readable somewhere.
+        assert before <= _contents(stack.seeded.lib, stack.archive), f"content lost at {barrier}"
+        # And nothing was removed that had not been verified first.
+        for operation in _plan_state(stack, plan["id"])["operations"]:
+            source = Path(operation["source_path"])
+            if not source.exists():
+                archived = Path(operation["destination_path"])
+                assert archived.exists(), f"{barrier}: source removed without an archive copy"
+                assert _sha256(archived) == operation["expected_sha256"]
+
+        # Recovery is offered from evidence, resolves without a worker, and converges.
+        verdicts = _recovery(stack)
+        assert all(v["classification"] != "manual" for v in verdicts["operations"]), verdicts
+        _resolve(stack)
+        assert _recovery(stack)["operations"] == []
+        # Repeating it changes nothing.
+        assert _resolve(stack) == {"resumed": 0, "completed": 0, "manual": 0}
+
+        # Recovery finishes what was durable and re-plans what was not. Before the
+        # archive copy existed, that means putting the item back to `planned` with the
+        # original untouched; from `verified` onwards it means completing it.
+        stack.restart_server()
+        operation = _plan_state(stack, plan["id"])["operations"][0]
+        asset = next(iter(stack.assets()))
+        if barrier == "transferring":
+            assert operation["journal_state"] == "planned"
+            assert Path(operation["source_path"]).exists()
+            assert asset["availability_state"] == "active"
+            assert _contents(stack.seeded.lib) == before
+        else:
+            assert operation["journal_state"] == "complete"
+            assert not Path(operation["source_path"]).exists()
+            assert _contents(stack.archive) == before
+            assert asset["availability_state"] == "archived_online"
+            assert _sha256(stack.archive / asset["archive_path"]) == asset["current_sha256"]
+            assert {entry["sha256"] for entry in _manifest(stack)} == before
+    finally:
+        stack.stop()
+
+
+def test_evidence_that_contradicts_the_journal_is_never_guessed(stack):
+    stack.start(worker=False)
+    location = stack.register()
+    plan = _plan(stack, location["id"])
+    # The journal claims a verified archive copy the medium does not have.
+    with session_factory(stack.seeded) as sf:
+        from photo_pipeline.services.archive_journal import ArchiveJournal
+
+        journal = ArchiveJournal(sf)
+        operation = journal.operations(plan["id"])[0]
+        journal.begin(operation["id"], worker_id="crashed", fencing_token=1)
+        journal.transition(operation["id"], "verified", fencing_token=1)
+
+    verdicts = _recovery(stack)
+    assert [v["classification"] for v in verdicts["operations"]] == ["manual"]
+    assert _resolve(stack)["manual"] == 1
+    # The source is untouched and the unresolved item keeps blocking a new archive.
+    assert (stack.seeded.lib / ALBUM / "a.jpg").exists()
+    blocked = _preflight(stack, location["id"])
+    assert "archive_pending" in {issue["code"] for issue in blocked["blockers"]}
+
+
+# ── offline identity and deduplication ───────────────────────────────────────
+
+
+def test_offline_photos_stay_hashed_previewable_and_deduplicable(stack):
+    stack.start()
+    result = _archive_album(stack)
+    archived = _assets(stack)
+    copy_source = stack.archive / next(iter(archived.values()))["archive_path"]
+    exact = stack.seeded.lib / "inbox" / "again.jpg"
+    exact.parent.mkdir(parents=True, exist_ok=True)
+    shutil.copy2(copy_source, exact)
+    stack.unmount()
+
+    scanned = httpx.post(f"{stack.base}/api/v1/inventory/scan", timeout=60)
+    scanned.raise_for_status()
+    # An unmounted medium is not a missing file.
+    assert scanned.json()["counts"].get("missing") is None
+    offline = {a["id"]: a for a in stack.assets() if a["id"] in archived}
+    assert {a["availability_state"] for a in offline.values()} == {"archived_offline"}
+    assert all(a["current_sha256"] for a in offline.values())
+
+    # The new active copy is recognised as the archived photo, not as a new one.
+    httpx.post(f"{stack.base}/api/v1/duplicates/detect", timeout=60).raise_for_status()
+    clusters = httpx.get(
+        f"{stack.base}/api/v1/duplicates/clusters", params={"limit": 50}, timeout=TIMEOUT
+    ).json()["items"]
+    exact_clusters = [c for c in clusters if c["method"] == "exact"]
+    assert len(exact_clusters) == 1
+    detail = httpx.get(
+        f"{stack.base}/api/v1/duplicates/clusters/{exact_clusters[0]['id']}", timeout=TIMEOUT
+    ).json()
+    canonical = next(m for m in detail["members"] if m["asset_id"] == detail["canonical_asset_id"])
+    assert canonical["availability_state"] == "archived_offline"
+    assert canonical["archive_location"] == result["location"]["name"]
+    assert canonical["preview"]["state"] == "ready" and canonical["preview"]["protected"]
+
+    # And the retained preview really is served while the medium is away.
+    thumbnail = httpx.get(
+        f"{stack.base}/api/v1/assets/{canonical['asset_id']}/thumbnail",
+        params={"size": 1280},
+        timeout=TIMEOUT,
+    )
+    assert thumbnail.status_code == 200
+    assert thumbnail.headers["content-type"] == "image/webp"
+
+
+def test_a_fuzzy_offline_match_asks_for_the_medium_instead_of_guessing(stack):
+    stack.start()
+    result = _archive_album(stack)
+    archived = next(iter(_assets(stack).values()))
+    variant = stack.seeded.lib / "inbox" / "resized.jpg"
+    variant.parent.mkdir(parents=True, exist_ok=True)
+    _resize(stack.archive / archived["archive_path"], variant)
+    stack.unmount()
+
+    httpx.post(f"{stack.base}/api/v1/inventory/scan", timeout=60).raise_for_status()
+    httpx.post(f"{stack.base}/api/v1/duplicates/detect", timeout=60).raise_for_status()
+    clusters = httpx.get(
+        f"{stack.base}/api/v1/duplicates/clusters", params={"limit": 50}, timeout=TIMEOUT
+    ).json()["items"]
+    perceptual = [c for c in clusters if c["method"] == "perceptual"]
+    assert len(perceptual) == 1
+
+    detail = httpx.get(
+        f"{stack.base}/api/v1/duplicates/clusters/{perceptual[0]['id']}", timeout=TIMEOUT
+    ).json()
+    # A fuzzy match is never decided automatically, and full-resolution review names
+    # the medium to mount rather than guessing from the preview.
+    assert detail["state"] == "open" and detail["requires_confirmation"] is True
+    assert detail["mount_required"] == [result["location"]["name"]]
+
+
+def test_mounting_the_medium_again_makes_the_originals_reachable(stack):
+    stack.start()
+    result = _archive_album(stack)
+    stack.unmount()
+    httpx.post(f"{stack.base}/api/v1/inventory/scan", timeout=60).raise_for_status()
+    assert {a["availability_state"] for a in stack.assets()} == {"archived_offline"}
+    # Restoring from a medium that is not there is refused, with its reason.
+    away = httpx.post(
+        f"{stack.base}/api/v1/restore-preflight",
+        json={"location_id": result["location"]["id"]},
+        timeout=TIMEOUT,
+    ).json()
+    assert away["state"] == "blocked"
+    assert "location_offline" in {issue["code"] for issue in away["blockers"]}
+
+    stack.remount()
+    locations = httpx.get(f"{stack.base}/api/v1/archive-locations", timeout=TIMEOUT).json()
+    assert locations["locations"][0]["state"] == "online"
+    assert {a["availability_state"] for a in stack.assets()} == {"archived_online"}
+
+
+# ── restore ──────────────────────────────────────────────────────────────────
+
+
+def test_restoring_returns_the_bytes_without_losing_identity_or_the_archive(stack):
+    stack.start()
+    result = _archive_album(stack)
+    identities = set(_assets(stack))
+    hashes = {a["id"]: a["current_sha256"] for a in stack.assets()}
+
+    _restore(stack, result["location"]["id"])
+
+    restored = _assets(stack)
+    assert set(restored) == identities  # the same photos, not new ones
+    for asset_id, sha256 in hashes.items():
+        asset = restored[asset_id]
+        assert asset["availability_state"] == "active"
+        assert _sha256(Path(asset["current_path"])) == sha256
+        # The archive copy is a copy: restoring empties nothing.
+        assert (stack.archive / asset["archive_path"]).exists()
+
+    stack.restart_server()
+    assert _assets(stack) == restored
+
+
+def test_a_restore_comes_back_beside_an_occupant_never_over_it(stack):
+    stack.start()
+    result = _archive_album(stack)
+    squatter = stack.seeded.lib / ALBUM / "a.jpg"
+    squatter.parent.mkdir(parents=True, exist_ok=True)
+    squatter.write_bytes(b"a different photo lives here now")
+
+    report = _restore(stack, result["location"]["id"])["preflight"]
+    destinations = {item["destination_path"] for item in report["items"]}
+    assert any("(restored)" in destination for destination in destinations)
+
+    assert squatter.read_bytes() == b"a different photo lives here now"
+    assert (stack.seeded.lib / ALBUM / "a (restored).jpg").exists()
+    assert {a["availability_state"] for a in stack.assets()} == {"active"}
+
+
+def _resize(source: Path, destination: Path, scale: float = 0.5) -> None:
+    from PIL import Image
+
+    with Image.open(source) as opened:
+        opened.resize(
+            (int(opened.width * scale), int(opened.height * scale)), Image.LANCZOS
+        ).save(destination, quality=95)

tests/e2e/test_traceability.py

@@ -14,6 +14,7 @@ MAP = json.loads((REPO / "tests" / "story_traceability.json").read_text())["stor
 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)}
 PHASE_E_STORIES = {f"US05-0{n}" for n in range(1, 7)}
+PHASE_F_STORIES = {f"US06-0{n}" for n in range(1, 7)}
 
 
 def test_all_phase_a_stories_are_mapped():
@@ -32,6 +33,13 @@ def test_all_phase_e_stories_are_mapped():
     assert PHASE_E_STORIES <= set(MAP)
 
 
+def test_all_phase_f_stories_are_mapped():
+    """US06-06 acceptance: every archive-lifecycle story, destination through
+    restore, is tied to automated tests — archiving is the only stage that removes
+    an original."""
+    assert PHASE_F_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

@@ -137,6 +137,12 @@
     ],
     "US06-04": [
       "tests/integration/test_restore.py"
+    ],
+    "US06-05": [
+      "tests/e2e/test_archive_ui.py"
+    ],
+    "US06-06": [
+      "tests/e2e/test_phase_f_pipeline.py"
     ]
   }
 }