US07-02: Harden API Authorization and Path Boundaries

AGENTS.md

@@ -80,7 +80,8 @@ are the authoritative backlog.
    uncertain. Work only on the claimed story and its generated feature branch.
 4. Read the entire issue, linked specification, dependencies, and acceptance criteria.
    Reconcile them with the concept before designing or changing code.
-5. Inspect the legacy CLI donors before replacing applicable behavior. Update the donor
+5. Inspect the legacy CLI donors — frozen in `legacy_cli_archive/` since US07-01,
+   with their ledger — before replacing applicable behavior. Update the donor
    ledger and characterization tests required by the story.
 6. Implement every acceptance criterion and its automated tests.
 7. Run story-specific tests and the accumulated regression suite required by the epic.

README.md

@@ -17,6 +17,25 @@ python -m photo_pipeline serve     # start the API + static review UI (127.0.0.1
 Configuration comes from `PHOTO_PIPELINE_*` environment variables (see
 `photo_pipeline/config.py`); secrets are referenced, never logged.
 
+### API access (US07-02)
+
+The app listens on loopback, so its attacker is another page in the same browser.
+Every `/api/v1` route except `health/live`, `health/ready`, and `session` requires
+the application session, and every mutation requires its CSRF token as well:
+
+```bash
+BASE=http://127.0.0.1:8000
+TOKEN=$(curl -sc /tmp/pp.jar $BASE/api/v1/session | python -c 'import json,sys; print(json.load(sys.stdin)["csrf_token"])')
+curl -sb /tmp/pp.jar -H "X-CSRF-Token: $TOKEN" -X POST $BASE/api/v1/albums/proposals -d '{}' -H 'Content-Type: application/json'
+```
+
+The session is per server process — restarting `serve` invalidates it, and the
+browser client re-bootstraps by itself. Requests are also refused when the `Host` is
+not a loopback name (DNS rebinding), when `Origin` is any other origin, when
+`Sec-Fetch-Site` says the request came from another site (an `<img>` pointed at a
+thumbnail), or when the body exceeds `PHOTO_PIPELINE_MAX_REQUEST_BYTES`. There is no
+CORS middleware at all, so no other origin can read a response.
+
 ## Testing
 
 One offline command runs the whole suite (unit, integration, and browser
@@ -169,3 +188,75 @@ 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.
+
+## Legacy CLI archive
+
+The command-line tools this application was extracted from are frozen in
+`legacy_cli_archive/` (US07-01): the original sources, their docs, the dependency
+lock they were last verified against, schema notes, a redacted sample
+configuration, the donor ledger, and a checksum for every file.
+
+```bash
+cd legacy_cli_archive && shasum -a 256 -c CHECKSUMS.sha256   # verify the archive
+work_item/scripts/python -m pytest tests/unit/test_legacy_archive.py -q   # lint it
+```
+
+They are reference material and rollback evidence only. No module under
+`photo_pipeline/` imports or executes them, the archive is not on the application's
+import path, and `tests/unit/test_legacy_archive.py` enforces that along with the
+checksums and the redaction. Only the two suites that compare *against* the donors —
+`tests/characterization/` and `tests/integration/test_safety_parity.py` — put the
+archived sources on `sys.path`.
+
+The last path-keyed state they owned, `nsfw_scores.csv`, is imported once and then
+left alone:
+
+```bash
+work_item/scripts/python -m photo_pipeline import-legacy-scores /path/to/nsfw_scores.csv --dry-run
+```
+
+The import writes scored-but-unreviewed `safety_reviews` rows onto stable asset ids,
+never invents an asset for an unknown path, never overwrites a human decision, and
+writes a reconciliation report to the data directory saying exactly what it did.
+`legacy_cli_archive/donor_ledger.yaml` records every migrated behavior with its
+target, the tests that pin the donor, the tests that prove the replacement, and each
+intentional delta; rows still marked `pending` name the story that will resolve them.

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

@@ -2,14 +2,43 @@
 // cancellation. Every method accepts an optional { signal } from cancellable().
 export const BASE = "/api/v1";
 
+// The API refuses every request without the session cookie, and every mutation
+// without this token echoed back. The token is readable only same-origin, which is
+// what makes it proof that the caller is this app and not another page.
+let csrfToken = null;
+
+async function session() {
+  if (csrfToken === null) {
+    const response = await fetch(BASE + "/session", { credentials: "same-origin" });
+    const body = await response.json().catch(() => null);
+    csrfToken = (body && body.csrf_token) || null;
+  }
+  return csrfToken || "";
+}
+
+async function send(path, { signal, ...options }) {
+  return fetch(BASE + path, {
+    credentials: "same-origin",
+    signal,
+    ...options,
+    headers: {
+      "Content-Type": "application/json",
+      "X-CSRF-Token": await session(),
+      ...(options.headers || {}),
+    },
+  });
+}
+
 async function request(path, { signal, ...options } = {}) {
   let response;
   try {
-    response = await fetch(BASE + path, {
-      headers: { "Content-Type": "application/json" },
-      signal,
-      ...options,
-    });
+    response = await send(path, { signal, ...options });
+    // A restarted server issues a new session; re-bootstrap once rather than
+    // stranding an open tab on 401.
+    if (response.status === 401) {
+      csrfToken = null;
+      response = await send(path, { signal, ...options });
+    }
   } catch (error) {
     // A caller-cancelled fetch is not a failure; tag it so views can ignore it.
     if (error.name === "AbortError") {
@@ -138,4 +167,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]}`;
+}

frontend/js/tests/unit.js

@@ -29,6 +29,10 @@ function jsonResponse(status, body) {
 const tick = (ms = 0) => new Promise((r) => setTimeout(r, ms));
 
 async function run() {
+  // The client fetches its CSRF token once, lazily (US07-02). Do that against the
+  // real server first, so the stubbed fetch below only ever sees the call under test.
+  await api.workflow().catch(() => {});
+
   // ── store ────────────────────────────────────────────────────────────────
   {
     const store = createStore({ n: 0 });

legacy_cli_archive/CHECKSUMS.sha256

@@ -0,0 +1,22 @@
+d098bbde13d2ebc872ca781e244cc1e48b97cc551fedaca3251d6a5278a15234  src/compare_models.py
+7d0f68cb95fbd6603e4c558620098b46929ebdb6fd91a598bbf84f26eea35e6c  src/nsfw_tag.py
+35329c53570e215cef15f59429c1a5251a0448b90522b0910ab9d68efe1bc307  src/nsfwtag/__init__.py
+7f52b55e4f3b77eda3657d7cd2272c7cca8422100e241671de495603c1126ee2  src/nsfwtag/__main__.py
+6f5a97114e0d87d272ce22d065dc31fa0dd72ff8590d11ad14cb1c1f486d339c  src/nsfwtag/bench.py
+29498fac1d73ba2b7420ffe1ffe49a87f684ec299e2c3b9c6e14e63e95643ba4  src/nsfwtag/exif.py
+934e82c402813ebf503e5a20eb03d5103f84df13117bc4dd1fd95deaa01ab263  src/nsfwtag/README.md
+68125e6184c7e4d2a5b0153f2155675753ab0b1329d1dbc4933d4769492ec4be  src/nsfwtag/review.html
+a597eab74803dd31452b5d70abf7d6d6320eb17a61596c891b009da15d624e98  src/nsfwtag/scoring.py
+674969a18e58ee511a2abf574a875db92f2f524769b1b614d0d3662ab879a709  src/nsfwtag/server.py
+24a7c8d029da6e97d46b110a9fe9dbb3900127f8ae3c142f634fdd91d6c45243  src/nsfwtag/webapp.py
+2c2ea558f1b9095c1150f078ea29c8e0f180cc045a9d1097b83f61bfb145295b  src/photo_analyzer.py
+5fb7ce2f977a17da1f501c45d7985328318ae14298a3f347eab46a6e3f02aab3  src/test_dedup.py
+1673dc76cc60aba56a5b065d9bc9f5dda00342cf2a79d000a3f74686fb7cffec  src/test_nsfw_skip.py
+55aa82f348e0e90be0163ba6aa278e5fc8e445996e5335774917e5ef59ad3563  src/webapp/__init__.py
+ba4342bd0175591a2121f063f14a78a602d5679bcfcf24aa14c8d4b8cd5da1a2  src/webapp/__main__.py
+a60f24035a989909467778f888d62854c1afc2702f47b8d3cca32bf3cf04d5cf  src/webapp/analyzer.html
+018bcc2f770d716444b456db58d6b4a41800138a98598b3c27ba503ab22d939e  src/webapp/page.py
+5f599b107b2b117ca118b6fbdec5e5786a9ab4eea424aa9af1bfea7bf87cebab  src/webapp/query.py
+875cce697caa02717c85a2707ba935c46dd9ace18428fd4abb13105b7a8e24d9  src/webapp/README.md
+af7d0d72d245b4bbb1e0e30f9708239697543212d4171defd1bfe26dd24a4f05  src/webapp/runner.py
+cddf0555b06fe7e4daa14309baaaf91130f8506922cb0ffb6507d76cbe39add4  src/webapp/server.py

legacy_cli_archive/README.md

@@ -0,0 +1,80 @@
+# Legacy CLI archive (US07-01)
+
+Frozen, read-only sources of the command-line tools this application was extracted
+from. They are **reference material and rollback evidence** — provenance for
+behavior that now lives in `photo_pipeline/`, and the only way to answer "what did
+the original actually do?" once the replacement has drifted.
+
+> **Nothing here is production code.** No module under `photo_pipeline/` imports or
+> executes anything in this directory, and this directory is not on the application's
+> import path. `tests/unit/test_legacy_archive.py` enforces both, along with the
+> checksums and the redaction below.
+
+## What is here
+
+| Path | Role |
+|---|---|
+| `src/photo_analyzer.py` | the analysis CLI: discovery, hashing/dedup, vision analysis, EXIF writing, SQLite schema, album naming |
+| `src/nsfwtag/` | NSFW scoring, EXIF safety keywords, and the review server (`__init__` 1.1.0) |
+| `src/webapp/` | the stdlib review web app: FTS search, stats, subprocess runner, HTML shell (`__init__` 0.1.0) |
+| `src/nsfw_tag.py` | thin backwards-compatible entry point for `nsfwtag` |
+| `src/compare_models.py` | dev-only model comparison script |
+| `src/test_dedup.py`, `src/test_nsfw_skip.py` | the CLIs' own standalone self-checks (never pytest suites) |
+| `donor_ledger.yaml` | the donor ledger: every migrated behavior, its target, its tests, and every intentional delta |
+| `requirements-lock.txt` | the dependency versions the frozen sources were last verified against |
+| `photo_analyzer.env.sample` | the CLI's configuration surface, with every value replaced by a placeholder |
+| `CHECKSUMS.sha256` | SHA-256 of every archived source file |
+
+`photo_analyzer.py` carries no `__version__`; its identity is its checksum, recorded
+in `CHECKSUMS.sha256` and taken at commit `9b7ee6b` (the merge of US06-06, the last
+commit before archival).
+
+## Verifying the archive
+
+```bash
+cd legacy_cli_archive && shasum -a 256 -c CHECKSUMS.sha256
+```
+
+Any edit to an archived source must be accompanied by a regenerated checksum file
+and a note here explaining why a *frozen* archive changed — the normal answer being
+that it should not.
+
+## Schema notes
+
+`photo_analyzer.py` owned a path-keyed SQLite database (`SCHEMA`, near the top of the
+file):
+
+- `photos(id, path UNIQUE, status, phash, file_sha1, dup_of, description, tags,
+  people_count, setting, time_of_day, season, mood, location_hint, approx_year,
+  raw_response, error_message, analyzed_at, exif_written_at)`;
+- `photos_fts` — an FTS5 external-content index over `path, description, tags, mood,
+  location_hint`, kept in sync by insert/update/delete triggers;
+- late columns (`phash`, `file_sha1`, `dup_of`) were added by an in-code
+  `_migrate_schema()` rather than a migration tool, and their indexes are created
+  only after the `ALTER`.
+
+The replacement keeps the same analysis fields but re-keys everything to a stable
+`assets.id` (Alembic migrations `0001`…), because a path is not an identity: the
+donor's `path UNIQUE` is exactly what broke on every move and rename.
+
+`nsfwtag` kept its safety scores outside the database in `nsfw_scores.csv`
+(`path,nsfw_score`, four decimals, unreadable rows dropped). That file is no longer a
+source of truth; `photo_pipeline/services/legacy_import.py` imports it into
+`assets.safety_score` and reports exactly what matched, what did not, and why.
+
+## Redaction
+
+The archive contains no credentials. `photo_analyzer.env.sample` documents the
+configuration surface (`LLM_API_KEY`, `LLM_BASE_URL`, `LLM_MODEL`, and the tuning
+variables) with placeholder values only; the CLI itself never contained a key, it
+read one from `photo_analyzer.env` or the environment. No `.env`, database, log, CSV,
+or photo from the author's library is archived.
+
+## Why these tools were retired
+
+Each behavior's fate is recorded per row in `donor_ledger.yaml`: `reuse`, `extract`,
+`refactor`, or `replace`, with the target module, the characterization tests that
+pinned the donor's behavior, the parity tests the replacement passes, and — where the
+replacement deliberately does something else or nothing at all — a `delta` saying so.
+Rows still marked `pending` name the backlog story that will resolve them; they are
+the honest list of what has *not* been carried over yet.

legacy_cli_archive/donor_ledger.yaml

@@ -12,7 +12,15 @@
 #                            (if anything) carries over
 # Every row needs either `tests` (existing test IDs, module::function) or
 # `pending_story` (the backlog story that will characterize/deliver it).
-# status: characterized | pending
+# status: characterized — donor behavior pinned by characterization tests
+#         resolved      — replacement shipped; `parity` names the tests that prove
+#                         it, and `delta` states every intentional difference
+#         pending       — not migrated yet; `pending_story` says which story will
+# parity: test ids (path::function) in any suite, proving the replacement
+# delta:  what the replacement deliberately does differently, or not at all
+#
+# Archived by US07-01: the sources referenced below now live beside this file in
+# src/ and are frozen (see README.md). Nothing in photo_pipeline imports them.
 
 rows:
   # ── photo_analyzer.py ──────────────────────────────────────────────────────
@@ -80,8 +88,12 @@ rows:
       Copying a primary's analysis into variant rows survives, but keyed by
       asset_id and recorded as stage state instead of raw row copies.
     target: photo_pipeline/services/duplicates.py
-    pending_story: US01-04
-    status: pending
+    parity:
+      - tests/integration/test_duplicate_engine.py::test_exact_copies_form_auto_decided_cluster
+      - tests/integration/test_duplicate_engine.py::test_perceptual_variant_is_review_only
+    delta: >
+      The donor propagated variant links implicitly while writing rows; the replacement makes the canonical link a reviewable, reversible cluster decision, so a propagated link can always be undone.
+    status: resolved
 
   - id: pa-hashing
     area: hashing
@@ -141,8 +153,12 @@ rows:
     classification: replace
     rationale: Console report; superseded by the duplicate-review API/UI (US01-06).
     target: photo_pipeline/api/routes + frontend duplicate review
-    pending_story: US01-06
-    status: pending
+    parity:
+      - tests/integration/test_review_api.py::test_clusters_list_and_detail
+      - tests/e2e/test_review_ui.py::test_fuzzy_decision_requires_confirmation
+    delta: >
+      The text listing became the paged cluster API and the comparison UI; no textual report is produced.
+    status: resolved
 
   - id: pa-reconcile
     area: database
@@ -202,7 +218,7 @@ rows:
       RGB-normalize (drops alpha, converts HEIC), LANCZOS resize to 2048px
       long-edge, JPEG q85 base64 — the provider-input contract. Truncated-image
       tolerance (ImageFile.LOAD_TRUNCATED_IMAGES) carries with it.
-    target: photo_pipeline/integrations/vision.py
+    target: photo_pipeline/services/analysis.py
     tests:
       - test_pa_imaging::test_prepare_image_small_passthrough_jpeg
       - test_pa_imaging::test_prepare_image_resizes_to_max_long_edge
@@ -230,9 +246,12 @@ rows:
       response validation, 429/503 retry with exponential backoff. Prompt and
       model/config version must be persisted per analysis_runs. Characterized
       against a deterministic fake provider when the analysis service is ported.
-    target: photo_pipeline/integrations/vision.py
-    pending_story: US02-06
-    status: pending
+    target: photo_pipeline/services/analysis.py
+    parity:
+      - tests/integration/test_safety_analysis.py::test_provider_called_only_for_confirmed_sfw
+    delta: >
+      The prompt and response schema carry over; the provider is an injected adapter so the privacy gate is testable, and results are keyed to asset ids rather than paths.
+    status: resolved
 
   - id: pa-throttle
     area: logging
@@ -241,8 +260,8 @@ rows:
     rationale: >
       Rolling throttle window + persistent throttle_events.jsonl + RPD day
       counter become job metrics/events on the durable job model.
-    target: photo_pipeline/jobs/coordinator.py
-    pending_story: US02-02
+    target: photo_pipeline/services/jobs.py + photo_pipeline/services/analysis.py
+    pending_story: US07-04
     status: pending
 
   - id: pa-nsfw-filter
@@ -312,8 +331,12 @@ rows:
       checked between items, double-SIGINT force quit — becomes the durable
       JobRunner worker loop with the same drain-and-resume semantics.
     target: photo_pipeline/jobs/worker.py
-    pending_story: US02-02
-    status: pending
+    parity:
+      - tests/integration/test_worker.py::test_worker_processes_all_items
+      - tests/integration/test_worker.py::test_cooperative_cancellation_leaves_items_resumable
+    delta: >
+      The in-process folder loop with SIGINT handling became durable jobs claimed by a worker: cancellation is a persisted request, not a signal, and an interrupted run resumes from the database instead of restarting.
+    status: resolved
 
   - id: pa-ui-terminal
     area: ui
@@ -371,8 +394,12 @@ rows:
       the JSONL history logger become structured JSON logging with job_id/
       asset_id and job_events rows; per-photo history maps to job events.
     target: photo_pipeline structured logging + jobs/job_events
-    pending_story: US02-02
-    status: pending
+    parity:
+      - tests/integration/test_jobs.py::test_enqueue_persists_items_and_event
+      - tests/integration/test_jobs_sse.py::test_sse_streams_all_events_then_closes
+    delta: >
+      The JSONL history file and rich console handler are replaced by structured JSON logs plus durable job_events; the browser reads events over SSE rather than tailing a file.
+    status: resolved
 
   - id: pa-balance
     area: vision
@@ -382,8 +409,11 @@ rows:
       Provider balance/quota probes (report 'unsupported' on providers without
       the endpoint). Network-bound; characterized against the fake provider.
     target: photo_pipeline/services/analysis.py
-    pending_story: US02-06
-    status: pending
+    parity:
+      - tests/integration/test_safety_analysis.py::test_provider_called_only_for_confirmed_sfw
+    delta: >
+      Not carried over: balance/quota polling was provider-specific (Gemini/OpenAI billing endpoints) and key-scoped. Cost reporting, when a story asks for it, comes from the per-run usage recorded with each analysis result rather than from a vendor endpoint.
+    status: resolved
 
   - id: pa-cli
     area: configuration
@@ -393,9 +423,13 @@ rows:
       argparse surface is superseded by the API; flags map to job configs
       (documented in WEBAPP_CONCEPT.md §8 parity table). Transitional CLI calls
       the shared services until archival (E07).
-    target: photo_pipeline/api + transitional CLI
-    pending_story: US07-01
-    status: pending
+    target: photo_pipeline/__main__.py (serve | worker | migrate | import-legacy-scores) + /api/v1
+    parity:
+      - tests/integration/test_app_lifecycle.py::test_restart_preserves_data_and_reruns_migrations
+      - tests/unit/test_legacy_archive.py::test_production_code_never_imports_an_archived_module
+    delta: >
+      The argparse surface is not reproduced. Every flag that drove work became an API command or a job configuration; the CLI keeps only what an application needs to be operated (serve, worker, migrate) plus the one-off legacy CSV import.
+    status: resolved
 
   # ── nsfwtag/ ───────────────────────────────────────────────────────────────
   - id: nt-discovery
@@ -422,9 +456,17 @@ rows:
       nsfw_scores.csv stops being the source of truth (concept: DB state).
       Format characterized (4-decimal scores, bad rows dropped) because the
       existing CSV must migrate into assets.safety_score.
-    target: photo_pipeline/repositories (safety), CSV import in US01-02 migration
+    target: photo_pipeline/services/legacy_import.py (one-off import into safety_reviews)
     tests: [test_nsfwtag::test_score_cache_roundtrip_and_tolerance]
-    status: characterized
+    parity:
+      - tests/integration/test_legacy_import.py::test_scores_are_imported_onto_asset_identity
+      - tests/integration/test_legacy_import.py::test_a_reviewed_asset_is_never_overwritten_by_the_csv
+    delta: >
+      The CSV is no longer read at runtime at all: it is imported once into
+      scored-but-unreviewed safety_reviews rows and left on disk untouched. A path
+      that matches nothing is reported, never turned into an asset, and a human
+      decision always outranks an imported score.
+    status: resolved
 
   - id: nt-score-model
     area: nsfw
@@ -485,8 +527,11 @@ rows:
       Newline-list bulk tagging (nsfw_confirmed.txt flow) is superseded by DB
       review decisions; the existing list is a one-time migration input.
     target: photo_pipeline/services/safety.py (decision import in US01-02)
-    pending_story: US01-02
-    status: pending
+    parity:
+      - tests/integration/test_safety_parity.py::test_extracted_marks_and_partition_match_donor
+    delta: >
+      Bulk keyword application from a file list is replaced by decisions against asset ids; the EXIF write itself is the extracted, read-back-verified one.
+    status: resolved
 
   - id: nt-ui
     area: ui
@@ -498,8 +543,11 @@ rows:
       threshold/score review flow, lightbox and keyboard model are the frontend
       donor for the Safety view (preserved per concept §10; ported in US02-01).
     target: photo_pipeline/api + frontend Safety view
-    pending_story: US02-01
-    status: pending
+    parity:
+      - tests/e2e/test_workflow_views.py::test_safety_review_decide_persists_across_reload
+    delta: >
+      The stdlib review server is replaced by the API plus the Safety view; the donor's layout, thresholds, and keyboard flow carry over, its embedded HTML generation does not.
+    status: resolved
 
   - id: nt-bench
     area: nsfw
@@ -509,8 +557,11 @@ rows:
       Dev-only model benchmark; archived without webapp replacement (recorded
       basis of the AdamCodd model choice). No production caller.
     target: none (archive as reference)
-    pending_story: US07-01
-    status: pending
+    parity:
+      - tests/unit/test_legacy_archive.py::test_every_archived_source_matches_its_checksum
+    delta: >
+      No replacement: a dev-only benchmark whose result (the AdamCodd model choice) is already recorded. Kept in the archive as the basis of that choice.
+    status: resolved
 
   # ── webapp/ ────────────────────────────────────────────────────────────────
   - id: wa-query-search
@@ -579,9 +630,13 @@ rows:
       Subprocess-driving-the-CLI job control is superseded by durable DB jobs
       with a worker process. Two ideas carry over: progress derived from DB
       counts (not job-private state) and single-mutating-job enforcement.
-    target: photo_pipeline/jobs/coordinator.py
-    pending_story: US02-02
-    status: pending
+    target: photo_pipeline/services/jobs.py + photo_pipeline/jobs/worker.py
+    parity:
+      - tests/integration/test_jobs.py::test_idempotency_key_returns_same_job
+      - tests/integration/test_worker.py::test_handler_failure_fails_the_job
+    delta: >
+      Subprocess supervision of a CLI is replaced by durable jobs in the same process family: there is no subprocess to supervise, and progress is persisted rather than scraped from stdout.
+    status: resolved
 
   - id: wa-server
     area: ui
@@ -593,5 +648,9 @@ rows:
       browser. analyzer.html + page.py design (dark OLED tokens, Library/
       Analyze/Stats views) is frontend donor material per concept §10.
     target: photo_pipeline/api/app.py + frontend
-    pending_story: US02-05
-    status: pending
+    parity:
+      - tests/e2e/test_frontend_shell.py::test_shell_loads_assets_without_console_or_network_errors
+      - tests/e2e/test_frontend_shell.py::test_deep_link_and_reload_restore_view_and_filters
+    delta: >
+      The stdlib HTTP routes become the versioned FastAPI API and the static frontend shell; Python no longer interpolates HTML.
+    status: resolved

legacy_cli_archive/photo_analyzer.env.sample

@@ -0,0 +1,18 @@
+# Sample configuration for the archived photo_analyzer.py CLI (US07-01).
+#
+# REDACTED: no value below is real. Each line names a variable the CLI read and
+# what belongs there; the placeholders are deliberately not key-shaped, so this
+# file can never be mistaken for — or scanned as — a credential.
+
+LLM_API_KEY=<paste your provider key here>
+LLM_BASE_URL=<provider base url, e.g. the OpenAI-compatible Gemini endpoint>
+LLM_MODEL=<model id, e.g. a Gemini Flash release>
+
+# Optional tuning the CLI read from the same file:
+PHASH_THRESHOLD=8
+MAX_WORKERS=4
+RETRY_ATTEMPTS=3
+RPD_LIMIT=0
+
+# The CLI ignored its own shipped placeholder (a literal "sk-REPLACE..." string)
+# until it was replaced, and a shell variable always won over this file.

legacy_cli_archive/requirements-lock.txt

@@ -0,0 +1,20 @@
+# Final dependency lock of the archived CLIs (US07-01).
+#
+# These are the versions present in the environment the archive was taken from —
+# what the frozen sources were last verified against by the characterization
+# suite. Restoring a donor for forensics means pinning these, not "latest".
+#
+# Python 3.14.6
+
+openai==3.0.0          # photo_analyzer: OpenAI-compatible vision client
+numpy==2.4.6           # photo_analyzer, nsfwtag: pixel work
+Pillow==12.3.0         # photo_analyzer, nsfwtag: decode/resize
+rich==15.0.0           # photo_analyzer: console output
+scipy==1.18.0          # photo_analyzer: perceptual-hash DCT
+PyYAML==6.0.3          # tooling that reads the donor ledger
+
+# NSFW inference (nsfwtag/scoring.py, nsfwtag/bench.py) was never installed in the
+# archiving environment; the model stack is recorded here from the sources so a
+# forensic run can reproduce it, not from a resolved lock:
+#   torch, transformers, timm  — AdamCodd/vit-base-nsfw-detector (see nsfwtag/README.md)
+# exiftool is an external binary, not a Python package.

migrations/versions/0014_restore_plans.py

@@ -0,0 +1,37 @@
+"""Restore plans and archive divergence (US06-04).
+
+Revision ID: 0014_restore_plans
+Revises: 0013_protected_thumbnails
+Create Date: 2026-08-16
+
+Restore reuses the archive plan and journal tables: the crash-safe question is the
+same one in the opposite direction (copy, verify, publish, register), so the rows
+gain a ``direction`` instead of a parallel pair of tables. ``archive_divergent_at``
+records the moment an archived copy was proven to hold bytes that are not the ones
+the database recorded — a restore must never silently accept a different file.
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = "0014_restore_plans"
+down_revision = "0013_protected_thumbnails"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    for table in ("archive_plans", "archive_operations"):
+        op.add_column(
+            table,
+            sa.Column("direction", sa.String(), nullable=False, server_default="archive"),
+        )
+    op.add_column(
+        "assets", sa.Column("archive_divergent_at", sa.DateTime(timezone=True), nullable=True)
+    )
+
+
+def downgrade() -> None:
+    op.drop_column("assets", "archive_divergent_at")
+    for table in ("archive_plans", "archive_operations"):
+        op.drop_column(table, "direction")

photo_pipeline/__main__.py

@@ -1,4 +1,4 @@
-"""Application management CLI: ``python -m photo_pipeline {serve,migrate}``."""
+"""Application management CLI: ``python -m photo_pipeline {serve,migrate,worker,import-legacy-scores}``."""
 
 from __future__ import annotations
 
@@ -16,6 +16,17 @@ def main(argv: Sequence[str] | None = None) -> int:
     commands.add_parser("migrate", help="Upgrade the database to the latest revision")
     worker_cmd = commands.add_parser("worker", help="Run a durable-job worker")
     worker_cmd.add_argument("--id", default="worker-1", help="Worker id (lease owner)")
+    import_cmd = commands.add_parser(
+        "import-legacy-scores",
+        help="Import the archived CLI's nsfw_scores.csv into the database (US07-01)",
+    )
+    import_cmd.add_argument("csv", help="Path to nsfw_scores.csv")
+    import_cmd.add_argument(
+        "--overwrite", action="store_true", help="Replace differing imported scores"
+    )
+    import_cmd.add_argument(
+        "--dry-run", action="store_true", help="Report what would happen and change nothing"
+    )
     args = parser.parse_args(argv)
 
     config = Config.from_env()
@@ -25,6 +36,24 @@ def main(argv: Sequence[str] | None = None) -> int:
         run_migrations(config.database_url)
         return 0
 
+    if args.command == "import-legacy-scores":
+        import json
+
+        from photo_pipeline.db import create_db_engine, create_session_factory
+        from photo_pipeline.services.legacy_import import LegacyImportService, write_report
+
+        run_migrations(config.database_url)
+        engine = create_db_engine(config.database_url)
+        service = LegacyImportService(create_session_factory(engine))
+        report = service.import_nsfw_scores(
+            args.csv, overwrite=args.overwrite, dry_run=args.dry_run
+        )
+        # The report is the point: an import nobody can audit is not a migration.
+        if not args.dry_run:
+            write_report(report, config.data_dir)
+        print(json.dumps(report.counts, indent=2))
+        return 0
+
     if args.command == "worker":
         from photo_pipeline.db import create_db_engine, create_session_factory
 

photo_pipeline/api/app.py

@@ -8,11 +8,15 @@ and exposes the versioned ``/api/v1`` surface; US01-02 ships only health.
 
 from __future__ import annotations
 
+import logging
 from contextlib import asynccontextmanager
 from pathlib import Path
 
-from fastapi import FastAPI
+from fastapi import FastAPI, Request
+from fastapi.exceptions import RequestValidationError
+from fastapi.responses import JSONResponse
 from fastapi.staticfiles import StaticFiles
+from starlette.exceptions import HTTPException as StarletteHTTPException
 
 from photo_pipeline.api.routes import (
     albums,
@@ -25,10 +29,12 @@ from photo_pipeline.api.routes import (
     library,
     renames,
     safety,
+    session as session_routes,
     thumbnails,
     uploads,
     workflow,
 )
+from photo_pipeline.api.security import DEFAULT_HEADERS, SecurityMiddleware, Session
 
 # Registers the safety_score / analysis job handlers on import.
 import photo_pipeline.jobs.domain_handlers  # noqa: F401
@@ -39,6 +45,41 @@ from photo_pipeline.services.upload_batches import UploadBatchService
 
 FRONTEND_DIR = Path(__file__).resolve().parents[2] / "frontend"
 
+log = logging.getLogger(__name__)
+
+
+def _envelope(status: int, code: str, message: str) -> JSONResponse:
+    return JSONResponse(
+        status_code=status,
+        content={"error": {"code": code, "message": message}},
+        headers=DEFAULT_HEADERS,
+    )
+
+
+def _install_error_handlers(app: FastAPI) -> None:
+    """One JSON error envelope everywhere, and nothing behind it.
+
+    An unhandled exception carries the library's absolute paths, SQL, and sometimes
+    a credential in its text; the client gets a code, the operator gets the traceback
+    in the server log (US07-02).
+    """
+
+    @app.exception_handler(StarletteHTTPException)
+    async def _http_error(request: Request, exc: StarletteHTTPException):
+        return _envelope(exc.status_code, "http_error", str(exc.detail))
+
+    @app.exception_handler(RequestValidationError)
+    async def _validation_error(request: Request, exc: RequestValidationError):
+        # Field locations only: the echoed input can be the caller's own data, but it
+        # is also what ends up in shared logs and screenshots.
+        fields = sorted(".".join(str(part) for part in error["loc"]) for error in exc.errors())
+        return _envelope(422, "invalid_request", f"invalid request fields: {', '.join(fields)}")
+
+    @app.exception_handler(Exception)
+    async def _unhandled(request: Request, exc: Exception):
+        log.exception("unhandled error serving %s", request.url.path)
+        return _envelope(500, "internal_error", "internal error")
+
 
 def create_app(config: Config | None = None) -> FastAPI:
     config = config or Config.from_env()
@@ -62,6 +103,12 @@ def create_app(config: Config | None = None) -> FastAPI:
             app.state.engine = None
 
     app = FastAPI(title="Photo Pipeline", version="0.1.0", lifespan=lifespan)
+    # One session per process: the browser exchanges it for a cookie + CSRF token,
+    # and every other origin is refused before a route ever runs (US07-02).
+    app.state.session = Session.create()
+    app.add_middleware(SecurityMiddleware, session=app.state.session, config=config)
+    _install_error_handlers(app)
+    app.include_router(session_routes.router, prefix="/api/v1")
     app.include_router(health.router, prefix="/api/v1")
     app.include_router(inventory.router, prefix="/api/v1")
     app.include_router(duplicates.router, prefix="/api/v1")

photo_pipeline/api/routes/analysis.py

@@ -18,7 +18,10 @@ router = APIRouter(tags=["analysis"])
 
 
 def _service(request: Request) -> AnalysisService:
-    return AnalysisService(request.app.state.session_factory)
+    return AnalysisService(
+        request.app.state.session_factory,
+        library_roots=tuple(request.app.state.config.library_roots),
+    )
 
 
 def _error(status: int, code: str, message: str) -> JSONResponse:

photo_pipeline/api/routes/archives.py

@@ -1,4 +1,4 @@
-"""Archive location, preflight, and plan API (US06-01, US06-02).
+"""Archive location, preflight, plan, and restore API (US06-01, US06-02, US06-04).
 
 Registering a location writes a marker onto the medium; preflight is a command
 rather than a read, because it probes the destination, hashes the scope, and issues
@@ -13,10 +13,11 @@ from fastapi import APIRouter, Request
 from fastapi.responses import JSONResponse
 from pydantic import BaseModel
 
-from photo_pipeline.jobs.domain_handlers import ARCHIVE_LOCK, ARCHIVE_PLAN
+from photo_pipeline.jobs.domain_handlers import ARCHIVE_LOCK, ARCHIVE_PLAN, RESTORE_PLAN
 from photo_pipeline.services.archives import ArchiveError, ArchiveService
 from photo_pipeline.services.archive_transfer import ArchiveTransferService
 from photo_pipeline.services.jobs import JobBlocked, JobService
+from photo_pipeline.services.restores import RestoreService
 
 router = APIRouter(tags=["archives"])
 
@@ -42,10 +43,24 @@ class CreatePlanRequest(PreflightRequest):
     token: str
 
 
+class RestoreRequest(BaseModel):
+    location_id: str
+    # ``None`` means every asset archived at this location.
+    asset_ids: list[str] | None = None
+
+
+class CreateRestoreRequest(RestoreRequest):
+    token: str
+
+
 def _service(request: Request) -> ArchiveService:
     return ArchiveService(request.app.state.session_factory, config=request.app.state.config)
 
 
+def _restores(request: Request) -> RestoreService:
+    return RestoreService(request.app.state.session_factory, config=request.app.state.config)
+
+
 def _transfers(request: Request) -> ArchiveTransferService:
     return ArchiveTransferService(
         request.app.state.session_factory, config=request.app.state.config
@@ -138,6 +153,76 @@ def apply_plan(plan_id: str, request: Request):
     return {"plan_id": plan_id, "job": job}
 
 
+@router.post("/restore-preflight")
+def restore_preflight(body: RestoreRequest, request: Request):
+    """Validate restoring archived assets back into the library. Nothing moves."""
+    try:
+        return _restores(request).preflight(body.location_id, body.asset_ids)
+    except ArchiveError as error:
+        return _error(error)
+
+
+@router.post("/restore-plans", status_code=201)
+def create_restore_plan(body: CreateRestoreRequest, request: Request):
+    try:
+        return _restores(request).create(body.location_id, body.asset_ids, token=body.token)
+    except ArchiveError as error:
+        return _error(error)
+
+
+@router.get("/restore-plans")
+def list_restore_plans(request: Request) -> dict:
+    return {"plans": _restores(request).list()}
+
+
+@router.get("/restore-plans/{plan_id}")
+def get_restore_plan(plan_id: str, request: Request):
+    plan = _restores(request).get(plan_id)
+    if plan is None:
+        return _error(ArchiveError("unknown_plan", f"unknown restore plan {plan_id}"))
+    return plan
+
+
+@router.post("/restore-plans/{plan_id}/apply")
+def apply_restore_plan(plan_id: str, request: Request):
+    """Queue the restore on the archiver lane — the same single lane as archiving,
+    because both move the same originals."""
+    service = _restores(request)
+    plan = service.get(plan_id)
+    if plan is None:
+        return _error(ArchiveError("unknown_plan", f"unknown restore plan {plan_id}"))
+    unresolved = [row for row in service.journal.incomplete() if row["plan_id"] != plan_id]
+    if unresolved:
+        return _error(
+            ArchiveError(
+                "archive_pending",
+                f"an unresolved archive operation ({unresolved[0]['id']}) must be recovered",
+            )
+        )
+    try:
+        job = JobService(request.app.state.session_factory).enqueue(
+            RESTORE_PLAN,
+            lock=ARCHIVE_LOCK,
+            idempotency_key=f"restore:{plan_id}:{plan['version']}",
+            items=[plan_id],
+        )
+    except JobBlocked as error:
+        return JSONResponse(
+            status_code=409, content={"error": {"code": error.code, "message": str(error)}}
+        )
+    return {"plan_id": plan_id, "job": job}
+
+
+@router.get("/restore-recovery")
+def restore_recovery_status(request: Request) -> dict:
+    return _restores(request).recovery_status()
+
+
+@router.post("/restore-recovery/resolve")
+def resolve_restore_recovery(request: Request) -> dict:
+    return _restores(request).recover()
+
+
 @router.get("/archive-recovery")
 def recovery_status(request: Request) -> dict:
     """What an interrupted transfer left behind, straight from journal + disk."""

photo_pipeline/api/routes/session.py

@@ -0,0 +1,30 @@
+"""Session bootstrap: the one endpoint reachable without a session (US07-02).
+
+It sets the ``HttpOnly``/``SameSite=Strict`` session cookie and returns the CSRF
+token in the body. A foreign page can call this — it just cannot read the answer,
+because the app sends no CORS headers — and the cookie it received is never attached
+to a request that foreign page initiates.
+"""
+
+from __future__ import annotations
+
+from fastapi import APIRouter, Request
+from fastapi.responses import JSONResponse
+
+from photo_pipeline.api.security import SESSION_COOKIE
+
+router = APIRouter(tags=["session"])
+
+
+@router.get("/session")
+def start_session(request: Request) -> JSONResponse:
+    session = request.app.state.session
+    response = JSONResponse({"csrf_token": session.csrf_token})
+    response.set_cookie(
+        SESSION_COOKIE,
+        session.id,
+        httponly=True,
+        samesite="strict",
+        path="/",
+    )
+    return response

photo_pipeline/api/routes/thumbnails.py

@@ -31,5 +31,7 @@ def get_thumbnail(asset_id: str, request: Request, size: int = Query(512)):
     return FileResponse(
         path,
         media_type="image/webp",
-        headers={"Cache-Control": "public, max-age=31536000, immutable"},
+        # private: the URL is versioned and immutable, but these bytes are the user's
+        # photos and must never sit in a shared cache (US07-02).
+        headers={"Cache-Control": "private, max-age=31536000, immutable"},
     )

photo_pipeline/api/security.py

@@ -0,0 +1,192 @@
+"""Local-web attack surface: session, CSRF, Origin/Host checks, default headers.
+
+The app binds to 127.0.0.1, so the attacker is not a remote client but another page
+in the user's browser (concept §15, "Local web attack"): any site can issue requests
+to ``http://127.0.0.1:8000`` and can embed ``<img src=...>`` against media endpoints.
+The defenses stack, because each one alone has a hole:
+
+* **Host** must be a loopback name — a DNS rebinding host that resolves to 127.0.0.1
+  passes the browser's origin rules but not this check.
+* **Origin**, when the browser sends one, must be this exact origin (scheme, host,
+  port). There is no CORS middleware at all, so a foreign page can never *read* a
+  response even if it manages to send a request.
+* **Sec-Fetch-Site** rejects cross-site loads that carry no Origin, which is what an
+  ``<img>`` or ``<script>`` against a media endpoint looks like.
+* A **session cookie** (``SameSite=Strict``, ``HttpOnly``) is required by every
+  ``/api/v1`` route except liveness/readiness and the bootstrap itself. Strict means
+  the browser never attaches it to a request another site initiated.
+* A **CSRF token** must be echoed in a header on every mutation. It is handed out
+  only in the bootstrap response body, which a foreign page cannot read (no CORS) —
+  so possessing it proves the caller is same-origin.
+
+``evaluate`` is a pure function over the request metadata: the whole policy is one
+table that a unit test can enumerate, and the middleware only applies its verdict.
+"""
+
+from __future__ import annotations
+
+import secrets
+from collections.abc import Mapping
+from dataclasses import dataclass
+from urllib.parse import urlsplit
+
+from starlette.datastructures import Headers, MutableHeaders
+from starlette.responses import JSONResponse
+
+SESSION_COOKIE = "pp_session"
+CSRF_HEADER = "x-csrf-token"
+API_PREFIX = "/api/v1"
+SAFE_METHODS = frozenset({"GET", "HEAD", "OPTIONS"})
+# Reachable without a session: liveness/readiness (an orchestrator has no cookie)
+# and the bootstrap that issues the session in the first place.
+PUBLIC_PATHS = frozenset(
+    {f"{API_PREFIX}/health/live", f"{API_PREFIX}/health/ready", f"{API_PREFIX}/session"}
+)
+LOOPBACK_HOSTS = frozenset({"127.0.0.1", "localhost", "::1", "[::1]"})
+
+# Applied to every response. No inline script/style is used by the frontend, so the
+# policy can stay strict; `frame-ancestors 'none'` and CORP keep other pages from
+# embedding the app or its thumbnails.
+DEFAULT_HEADERS = {
+    "x-content-type-options": "nosniff",
+    "x-frame-options": "DENY",
+    "referrer-policy": "no-referrer",
+    "cross-origin-resource-policy": "same-origin",
+    "cross-origin-opener-policy": "same-origin",
+    "content-security-policy": (
+        "default-src 'self'; img-src 'self' data:; style-src 'self'; script-src 'self'; "
+        "connect-src 'self'; frame-ancestors 'none'; base-uri 'none'; form-action 'none'"
+    ),
+}
+
+
+@dataclass(frozen=True)
+class Session:
+    """One process, one session. A local app has exactly one user; a session store
+    would be bookkeeping without a second subject to distinguish.
+
+    ponytail: per-session rows if the app ever serves more than one operator.
+    """
+
+    id: str
+    csrf_token: str
+
+    @classmethod
+    def create(cls) -> Session:
+        return cls(secrets.token_urlsafe(32), secrets.token_urlsafe(32))
+
+
+@dataclass(frozen=True)
+class Refusal:
+    status: int
+    code: str
+    message: str
+
+
+def split_host(value: str) -> tuple[str, str]:
+    """``"127.0.0.1:8000"`` -> ``("127.0.0.1", "8000")``; bracketed IPv6 aware."""
+    value = value.strip()
+    if value.startswith("["):
+        host, _, port = value.partition("]")
+        return host + "]", port.lstrip(":")
+    host, _, port = value.partition(":")
+    return host, port
+
+
+def evaluate(
+    *,
+    method: str,
+    path: str,
+    headers: Mapping[str, str],
+    session: Session,
+    allowed_hosts: frozenset[str] = LOOPBACK_HOSTS,
+    max_request_bytes: int,
+) -> Refusal | None:
+    """Why this request must be refused, or ``None`` when it may proceed."""
+    host_header = headers.get("host", "")
+    host, port = split_host(host_header)
+    if host.lower() not in allowed_hosts:
+        return Refusal(403, "host_not_allowed", "request host is not a local address")
+
+    origin = headers.get("origin")
+    if origin is not None and origin != "":
+        parts = urlsplit(origin)
+        origin_host, origin_port = split_host(parts.netloc)
+        if (
+            parts.scheme not in ("http", "https")
+            or origin_host.lower() not in allowed_hosts
+            or origin_port != port
+        ):
+            return Refusal(403, "origin_not_allowed", "request origin is not this application")
+
+    # Absent means a non-browser client; "none" is a user-initiated navigation.
+    fetch_site = headers.get("sec-fetch-site")
+    if fetch_site is not None and fetch_site not in ("same-origin", "none"):
+        return Refusal(403, "cross_site_blocked", "cross-site requests are not accepted")
+
+    length = headers.get("content-length")
+    if length and length.isdigit() and int(length) > max_request_bytes:
+        return Refusal(413, "payload_too_large", "request body exceeds the configured limit")
+
+    protected = path.startswith(API_PREFIX) and path not in PUBLIC_PATHS
+    if not protected:
+        return None
+
+    if headers.get("cookie-session") != session.id:
+        return Refusal(401, "unauthenticated", "a valid application session is required")
+    if method.upper() not in SAFE_METHODS and headers.get(CSRF_HEADER) != session.csrf_token:
+        return Refusal(403, "csrf_failed", "missing or invalid CSRF token")
+    return None
+
+
+class SecurityMiddleware:
+    """Pure-ASGI so the SSE stream keeps streaming (BaseHTTPMiddleware buffers)."""
+
+    def __init__(self, app, *, session: Session, config) -> None:
+        self.app = app
+        self.session = session
+        self.max_request_bytes = config.max_request_bytes
+        self.allowed_hosts = frozenset(LOOPBACK_HOSTS | {str(config.host).lower()})
+
+    async def __call__(self, scope, receive, send) -> None:
+        if scope["type"] != "http":
+            await self.app(scope, receive, send)
+            return
+        headers = Headers(scope=scope)
+        # The cookie is read here and handed to the pure policy as one value, so the
+        # policy never has to parse a Cookie header.
+        lookup = dict(headers)
+        lookup["cookie-session"] = _cookie(headers.get("cookie", ""), SESSION_COOKIE)
+        refusal = evaluate(
+            method=scope.get("method", "GET"),
+            path=scope.get("path", "/"),
+            headers=lookup,
+            session=self.session,
+            allowed_hosts=self.allowed_hosts,
+            max_request_bytes=self.max_request_bytes,
+        )
+        if refusal is not None:
+            response = JSONResponse(
+                status_code=refusal.status,
+                content={"error": {"code": refusal.code, "message": refusal.message}},
+                headers=DEFAULT_HEADERS,
+            )
+            await response(scope, receive, send)
+            return
+
+        async def send_with_headers(message):
+            if message["type"] == "http.response.start":
+                out = MutableHeaders(scope=message)
+                for name, value in DEFAULT_HEADERS.items():
+                    out.setdefault(name, value)
+            await send(message)
+
+        await self.app(scope, receive, send_with_headers)
+
+
+def _cookie(header: str, name: str) -> str:
+    for part in header.split(";"):
+        key, _, value = part.strip().partition("=")
+        if key == name:
+            return value
+    return ""

photo_pipeline/config.py

@@ -30,6 +30,10 @@ class Config(BaseModel):
     log_level: str = "INFO"
     log_format: str = "json"  # "json" or "text"
 
+    # Largest request body the API accepts. Every endpoint takes small JSON commands;
+    # anything larger is a mistake or an attempt to exhaust memory (US07-02).
+    max_request_bytes: int = 1_048_576
+
     # Library boundary for path validation (os.pathsep-separated in the env var).
     library_roots: tuple[Path, ...] = ()
     thumbnail_cache_quota_bytes: int = 500_000_000

photo_pipeline/jobs/domain_handlers.py

@@ -1,9 +1,9 @@
 """Domain job handlers: safety scoring, content analysis, uploads, archive
-transfers (US02-06, US05-02, US06-02).
+transfers, restores (US02-06, US05-02, US06-02, US06-04).
 
 Importing this module registers the ``safety_score``, ``analysis``,
-``upload_batch``, and ``archive_plan`` job types so the generic worker can run them
-per item. Each handler delegates to its service, which owns the real work and the
+``upload_batch``, ``archive_plan``, and ``restore_plan`` job types so the generic
+worker can run them per item. Each handler delegates to its service, which owns the real work and the
 privacy gate. Handlers are idempotent: re-scoring or re-analyzing one asset is safe
 after an interrupted attempt, an upload batch refuses to re-run an attempt whose
 outcome is unknown, and an archive plan skips items it already completed.
@@ -20,6 +20,7 @@ SAFETY_SCORE = "safety_score"
 ANALYSIS = "analysis"
 UPLOAD_BATCH = "upload_batch"
 ARCHIVE_PLAN = "archive_plan"
+RESTORE_PLAN = "restore_plan"
 # Both mutate the library's metadata/derived state; one at a time (concept §one job).
 LIBRARY_WRITE_LOCK = "library_write"
 # The uploader lane: one album batch at a time (concept §16).
@@ -37,7 +38,8 @@ def _safety_score_item(asset_id: str, ctx: JobContext) -> None:
 def _analysis_item(asset_id: str, ctx: JobContext) -> None:
     from photo_pipeline.services.analysis import AnalysisService
 
-    AnalysisService(ctx.session_factory).run([asset_id])
+    roots = tuple(getattr(ctx.config, "library_roots", ()) or ())
+    AnalysisService(ctx.session_factory, library_roots=roots).run([asset_id])
 
 
 def _upload_batch_item(batch_id: str, ctx: JobContext) -> None:
@@ -70,7 +72,22 @@ def _archive_plan_item(plan_id: str, ctx: JobContext) -> None:
         raise RuntimeError(f"archive plan {plan_id}: {result['failed']} item(s) failed")
 
 
+def _restore_plan_item(plan_id: str, ctx: JobContext) -> None:
+    """One item = one restore plan. A restore removes nothing, so an item failure
+    simply leaves that asset archived (US06-04)."""
+    from photo_pipeline.config import Config
+    from photo_pipeline.services.restores import RestoreService
+
+    config = ctx.config if ctx.config is not None else Config.from_env()
+    result = RestoreService(ctx.session_factory, config=config).apply(
+        plan_id, worker_id=ctx.worker_id
+    )
+    if result["failed"]:
+        raise RuntimeError(f"restore plan {plan_id}: {result['failed']} item(s) failed")
+
+
 register(SAFETY_SCORE, _safety_score_item)
 register(ANALYSIS, _analysis_item)
 register(UPLOAD_BATCH, _upload_batch_item)
 register(ARCHIVE_PLAN, _archive_plan_item)
+register(RESTORE_PLAN, _restore_plan_item)

photo_pipeline/models/archives.py

@@ -66,6 +66,8 @@ class ArchivePlan(Base):
     # The preflight token this plan was approved against; re-verified before apply.
     token: Mapped[str] = mapped_column(String, nullable=False)
     albums: Mapped[str | None] = mapped_column(String)  # JSON array
+    # archive | restore — the same journal read in the opposite direction (US06-04).
+    direction: Mapped[str] = mapped_column(String, nullable=False, default="archive")
 
     # planned | applying | complete | failed
     state: Mapped[str] = mapped_column(String, nullable=False, default="planned")
@@ -100,6 +102,9 @@ class ArchiveOperation(Base):
     album: Mapped[str] = mapped_column(String, nullable=False)
     asset_id: Mapped[str] = mapped_column(ForeignKey("assets.id"), nullable=False, index=True)
 
+    # archive: library → medium. restore: medium → library (US06-04). ``source_path``
+    # and ``destination_path`` always mean "from" and "to" for this direction.
+    direction: Mapped[str] = mapped_column(String, nullable=False, default="archive")
     source_path: Mapped[str] = mapped_column(String, nullable=False)
     destination_path: Mapped[str] = mapped_column(String, nullable=False)
     # Relative to the location root, because the medium can be mounted elsewhere.

photo_pipeline/models/assets.py

@@ -43,6 +43,10 @@ class Asset(Base):
     # link is written and read by the archive service (US06-02).
     archive_location_id: Mapped[str | None] = mapped_column(String)
     archive_path: Mapped[str | None] = mapped_column(String)
+    # Set when the archived copy was proven to hold bytes other than the recorded
+    # ones (US06-04). Restore refuses such an asset instead of accepting a different
+    # file; cleared as soon as a verification matches again.
+    archive_divergent_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
     # Duplicate canonical link: NULL when the asset is itself canonical or undecided.
     canonical_asset_id: Mapped[str | None] = mapped_column(ForeignKey("assets.id"))
 

photo_pipeline/path_policy.py

@@ -49,6 +49,29 @@ def resolve_within(root: Path, path: os.PathLike | str) -> Path:
     return resolved
 
 
+def resolve_in_roots(roots: Iterable[os.PathLike | str], path: os.PathLike | str) -> Path:
+    """The resolved path, proven to be inside one of ``roots`` and not excluded.
+
+    Callers must use the **returned** path for whatever they do next: validating one
+    name and then opening another is the symlink race this exists to close (US07-02).
+    The message names no path — it reaches API responses.
+
+    With no roots configured there is no boundary to check; that is a property of the
+    configuration, not permission granted to this call.
+    """
+    if is_excluded(path):
+        raise PathPolicyError("path is inside an excluded (_IGNORE/) tree")
+    roots = list(roots)
+    if not roots:
+        return Path(path)
+    for root in roots:
+        try:
+            return resolve_within(Path(root), path)
+        except PathPolicyError:
+            continue
+    raise PathPolicyError("path is outside the configured library roots")
+
+
 def iter_supported_files(root: os.PathLike | str) -> Iterator[Path]:
     """Yield supported, non-excluded files under ``root`` in deterministic order.
 

photo_pipeline/services/analysis.py

@@ -25,6 +25,7 @@ from typing import Protocol
 from sqlalchemy import func, select
 from sqlalchemy.orm import sessionmaker
 
+from photo_pipeline import path_policy
 from photo_pipeline.integrations import exiftool
 from photo_pipeline.models import AnalysisResult, Asset, SafetyReview
 from photo_pipeline.services import hashing
@@ -59,9 +60,16 @@ def _now() -> datetime:
 
 
 class AnalysisService:
-    def __init__(self, session_factory: sessionmaker, *, provider: VisionProvider | None = None) -> None:
+    def __init__(
+        self,
+        session_factory: sessionmaker,
+        *,
+        provider: VisionProvider | None = None,
+        library_roots: tuple = (),
+    ) -> None:
         self._session_factory = session_factory
         self._provider = provider
+        self._roots = tuple(library_roots)
 
     def _sfw_asset_ids(self, session) -> set[str]:
         """Asset ids whose latest safety decision is ``sfw`` — the ONLY assets that
@@ -136,6 +144,18 @@ class AnalysisService:
             if not path:
                 skipped += 1
                 continue
+            # Second gate, at the moment of use: the database says where the file
+            # was, the filesystem decides what that name means now. A link swapped
+            # under an asset after the scan would otherwise send bytes from outside
+            # the library — the one place that leaves this machine (US07-02).
+            try:
+                path = str(path_policy.resolve_in_roots(self._roots, path))
+            except path_policy.PathPolicyError as error:
+                self._store(
+                    asset_id, status="error", result=None, error=str(error), tokens=0, raw=""
+                )
+                errors += 1
+                continue
             try:
                 result = provider.analyze(path, album_hint=_album_hint(path))
             except Exception as error:  # provider/validation failure is per-asset

photo_pipeline/services/archive_journal.py

@@ -15,6 +15,11 @@ planned → transferring → verified → removing → complete
               ↘             ↘          ↘ failed
 ```
 
+A restore (US06-04) uses the same rows with ``direction='restore'``: it copies from
+the medium back into the library and removes nothing, so it goes ``verified →
+complete`` directly. ``source_path``/``destination_path`` always mean "from"/"to",
+which is why the evidence table below needs no direction of its own.
+
 - ``transferring`` — intent recorded; a temporary copy may exist, the destination
   may or may not have been published. Nothing has been removed.
 - ``verified``     — the archived bytes exist at their final path, hash exactly as
@@ -73,10 +78,22 @@ ALLOWED_TRANSITIONS = {
     ArchiveState.FAILED: {ArchiveState.PLANNED, ArchiveState.TRANSFERRING},
 }
 
+# A restore removes nothing, so it has no ``removing`` step: a verified published
+# copy is the whole job (US06-04). Keeping this as a separate table means the
+# archive direction still cannot reach ``complete`` without going through removal.
+RESTORE_TRANSITIONS = {
+    **ALLOWED_TRANSITIONS,
+    ArchiveState.VERIFIED: {ArchiveState.COMPLETE, ArchiveState.FAILED},
+}
+
 TERMINAL_STATES = frozenset({ArchiveState.COMPLETE})
 # States where this item may already have touched the filesystem.
 UNSAFE_STATES = frozenset({ArchiveState.TRANSFERRING, ArchiveState.VERIFIED, ArchiveState.REMOVING})
 
+# Which way the bytes move. Same rows, same evidence table, opposite direction.
+ARCHIVE = "archive"
+RESTORE = "restore"
+
 RESUMABLE = "resumable"
 FORWARD = "forward"
 MANUAL = "manual"
@@ -94,8 +111,9 @@ class JournalConflict(JournalError):
     """Fencing check failed; a newer owner has taken over this operation."""
 
 
-def can_transition(current: str, target: str) -> bool:
-    return target in ALLOWED_TRANSITIONS.get(current, set())
+def can_transition(current: str, target: str, direction: str = ARCHIVE) -> bool:
+    table = RESTORE_TRANSITIONS if direction == RESTORE else ALLOWED_TRANSITIONS
+    return target in table.get(current, set())
 
 
 def _now() -> datetime:
@@ -119,7 +137,7 @@ class ArchiveJournal:
             if row.journal_state in TERMINAL_STATES:
                 raise InvalidTransition(f"{row.journal_state} is terminal")
             if row.journal_state != ArchiveState.TRANSFERRING and not can_transition(
-                row.journal_state, ArchiveState.TRANSFERRING
+                row.journal_state, ArchiveState.TRANSFERRING, row.direction
             ):
                 raise InvalidTransition(f"{row.journal_state} -> {ArchiveState.TRANSFERRING}")
             if row.journal_state != ArchiveState.TRANSFERRING:
@@ -160,7 +178,7 @@ class ArchiveJournal:
             if row.journal_state == target:
                 session.commit()
                 return _operation_dict(row)  # idempotent
-            if not can_transition(row.journal_state, target):
+            if not can_transition(row.journal_state, target, row.direction):
                 raise InvalidTransition(f"{row.journal_state} -> {target}")
 
             row.journal_state = target
@@ -192,16 +210,18 @@ class ArchiveJournal:
             )
             return [_operation_dict(row) for row in rows]
 
-    def incomplete(self) -> list[dict]:
+    def incomplete(self, *, direction: str | None = None) -> list[dict]:
         """Every operation left in a non-terminal, non-planned state — the work a
-        restart has to reason about."""
+        restart has to reason about. Without ``direction`` this spans archives and
+        restores, because either one half-done blocks the other."""
         with self._session_factory() as session:
+            stmt = select(ArchiveOperation).where(
+                ArchiveOperation.journal_state.not_in([*TERMINAL_STATES, ArchiveState.PLANNED])
+            )
+            if direction is not None:
+                stmt = stmt.where(ArchiveOperation.direction == direction)
             rows = session.scalars(
-                select(ArchiveOperation)
-                .where(
-                    ArchiveOperation.journal_state.not_in([*TERMINAL_STATES, ArchiveState.PLANNED])
-                )
-                .order_by(ArchiveOperation.plan_id, ArchiveOperation.sequence)
+                stmt.order_by(ArchiveOperation.plan_id, ArchiveOperation.sequence)
             )
             return [_operation_dict(row) for row in rows]
 
@@ -231,6 +251,7 @@ class ArchiveJournal:
         return {
             "operation_id": operation_id,
             "plan_id": row["plan_id"],
+            "direction": row["direction"],
             "album": row["album"],
             "asset_id": row["asset_id"],
             "source_path": row["source_path"],
@@ -243,8 +264,8 @@ class ArchiveJournal:
             "destination_matches": destination_matches,
         }
 
-    def classify_all(self) -> list[dict]:
-        return [self.classify(row["id"]) for row in self.incomplete()]
+    def classify_all(self, *, direction: str | None = None) -> list[dict]:
+        return [self.classify(row["id"]) for row in self.incomplete(direction=direction)]
 
     def blocks_mutation(self) -> bool:
         """True when any item may have the library half-archived."""
@@ -317,6 +338,7 @@ def _operation_dict(row: ArchiveOperation) -> dict:
     return {
         "id": row.id,
         "plan_id": row.plan_id,
+        "direction": row.direction,
         "sequence": row.sequence,
         "album": row.album,
         "asset_id": row.asset_id,

photo_pipeline/services/archive_transfer.py

@@ -54,6 +54,7 @@ from sqlalchemy.orm import sessionmaker
 from photo_pipeline.config import Config
 from photo_pipeline.models import ArchiveLocation, ArchiveOperation, ArchivePlan, Asset, AssetPath
 from photo_pipeline.services.archive_journal import (
+    ARCHIVE,
     MANUAL,
     RESUMABLE,
     ArchiveJournal,
@@ -115,6 +116,7 @@ class ArchiveTransferService:
                     location_id=location_id,
                     token=token,
                     albums=json.dumps(albums) if albums is not None else None,
+                    direction=ARCHIVE,
                     state="planned",
                     schema_version=MANIFEST_VERSION,
                     asset_count=preflight["totals"]["assets"],
@@ -132,6 +134,7 @@ class ArchiveTransferService:
                         ArchiveOperation(
                             id=str(uuid.uuid4()),
                             plan_id=plan_id,
+                            direction=ARCHIVE,
                             sequence=sequence,
                             album=album["album"],
                             asset_id=asset["asset_id"],
@@ -161,7 +164,11 @@ class ArchiveTransferService:
 
     def list(self) -> list[dict]:
         with self._session_factory() as session:
-            rows = session.scalars(select(ArchivePlan).order_by(ArchivePlan.created_at))
+            rows = session.scalars(
+                select(ArchivePlan)
+                .where(ArchivePlan.direction == ARCHIVE)
+                .order_by(ArchivePlan.created_at)
+            )
             return [_plan_dict(row) for row in rows]
 
     # ── apply ─────────────────────────────────────────────────────────────────
@@ -262,7 +269,7 @@ class ArchiveTransferService:
         if same_filesystem:
             os.rename(source, destination)
         else:
-            self._copy_and_publish(operation, source, destination)
+            copy_verify_publish(source, destination, operation["expected_sha256"])
         _fsync_dir(destination.parent)
 
         # 4. The published file is the archive only once it hashes as recorded.
@@ -282,27 +289,6 @@ class ArchiveTransferService:
         # 5. Only now may the active source go.
         self._finish(self.journal.get(operation["id"]), location, token=token, worker_id=worker_id)
 
-    def _copy_and_publish(self, operation: dict, source: Path, destination: Path) -> None:
-        """Cross-filesystem: copy to a temporary file beside the destination, prove
-        its bytes, then publish it atomically. The source is still untouched."""
-        temp = destination.with_name(f"{TEMP_PREFIX}{uuid.uuid4().hex}{TEMP_SUFFIX}")
-        try:
-            with open(source, "rb") as src, open(temp, "wb") as out:
-                shutil.copyfileobj(src, out, 1024 * 1024)
-                out.flush()
-                os.fsync(out.fileno())
-            if sha256_file(temp) != operation["expected_sha256"]:
-                raise PreconditionFailed("copy_mismatch", f"{source} copied with wrong bytes")
-            if destination.exists():
-                raise PreconditionFailed(
-                    "destination_exists", f"{destination} appeared during the transfer"
-                )
-            # ponytail: rename after an exists() check. The archiver lane is single
-            # and local; use O_EXCL/link-based publish if a second writer ever exists.
-            os.rename(temp, destination)
-        finally:
-            temp.unlink(missing_ok=True)
-
     def _finish(self, operation: dict, location: dict, *, token: int, worker_id: str) -> None:
         """Drive an item whose archive copy is durable through removal and
         bookkeeping. Every step is idempotent, so recovery may replay it."""
@@ -457,7 +443,7 @@ class ArchiveTransferService:
         """
         results = {"resumed": 0, "completed": 0, "manual": 0}
         touched: set[str] = set()
-        for verdict in self.journal.classify_all():
+        for verdict in self.journal.classify_all(direction=ARCHIVE):
             operation = self.journal.get(verdict["operation_id"])
             touched.add(operation["plan_id"])
             token = (operation["fencing_token"] or 0) + 1
@@ -484,7 +470,7 @@ class ArchiveTransferService:
         return results
 
     def recovery_status(self) -> dict:
-        verdicts = self.journal.classify_all()
+        verdicts = self.journal.classify_all(direction=ARCHIVE)
         return {
             "operations": verdicts,
             "manual": [v for v in verdicts if v["classification"] == MANUAL],
@@ -528,6 +514,33 @@ class ArchiveTransferService:
 # ── module helpers ───────────────────────────────────────────────────────────
 
 
+def copy_verify_publish(source: Path, destination: Path, expected_sha256: str) -> None:
+    """Copy to a temporary file beside the destination, prove its bytes, then publish
+    it atomically. The source is never touched, so a failure costs nothing.
+
+    Shared by archiving (library → medium) and restoring (medium → library, US06-04):
+    both need the same promise that a published file is either complete and correct
+    or not there at all.
+    """
+    temp = destination.with_name(f"{TEMP_PREFIX}{uuid.uuid4().hex}{TEMP_SUFFIX}")
+    try:
+        with open(source, "rb") as src, open(temp, "wb") as out:
+            shutil.copyfileobj(src, out, 1024 * 1024)
+            out.flush()
+            os.fsync(out.fileno())
+        if sha256_file(temp) != expected_sha256:
+            raise PreconditionFailed("copy_mismatch", f"{source} copied with wrong bytes")
+        if destination.exists():
+            raise PreconditionFailed(
+                "destination_exists", f"{destination} appeared during the transfer"
+            )
+        # ponytail: rename after an exists() check. The archiver lane is single and
+        # local; use O_EXCL/link-based publish if a second writer ever exists.
+        os.rename(temp, destination)
+    finally:
+        temp.unlink(missing_ok=True)
+
+
 def _same_filesystem(source: Path, destination_dir: Path) -> bool:
     """Proven at run time from the actual devices, never from the plan's preview."""
     try:
@@ -618,6 +631,7 @@ def _plan_dict(plan: ArchivePlan) -> dict:
         "id": plan.id,
         "location_id": plan.location_id,
         "token": plan.token,
+        "direction": plan.direction,
         "albums": json.loads(plan.albums) if plan.albums else None,
         "state": plan.state,
         "schema_version": plan.schema_version,

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"{report['capacity']['free_bytes']} B of free space",
+                    f"{self._config.archive_free_space_reserve_bytes} B reserve do not fit on "
+                    "the medium",
                 )
             )
         report["backup"] = self._backup_probe()

photo_pipeline/services/legacy_import.py

@@ -0,0 +1,220 @@
+"""Import the last path-keyed CSV state into the database (US07-01).
+
+``nsfwtag`` cached its safety scores in ``nsfw_scores.csv`` next to the library:
+one ``path,nsfw_score`` row per photo, four decimals, unreadable rows dropped. That
+file stops being a source of truth when the CLI is archived, so its scores are
+imported once — as scored-but-unreviewed ``safety_reviews`` rows on the stable
+``assets.id`` each path resolves to — and the CSV is left untouched on disk as its
+own backup.
+
+The import is deliberately conservative, because a score is evidence about a photo
+and a path is not an identity:
+
+- a row whose path matches no known asset is **unmatched**, never a new asset;
+- an asset a human already reviewed is **never** touched: a score is evidence, a
+  decision is a judgement, and an import may not overwrite the second with the
+  first — the difference is reported as a conflict instead;
+- a row for an asset that already carries a score is **skipped** unless
+  ``overwrite`` is asked for, and a differing score is reported either way;
+- an unparsable score or a duplicate path is **rejected** with its reason;
+- rerunning changes nothing that is already imported.
+
+Everything it did — and everything it refused to do — comes back as a
+reconciliation report, which the caller can persist next to the database. The
+donor's own reader is the specification for the format (donor ledger:
+``nt-score-cache``).
+"""
+
+from __future__ import annotations
+
+import csv
+import json
+import uuid
+from collections import Counter
+from dataclasses import asdict, dataclass, field
+from datetime import datetime, timezone
+from pathlib import Path
+
+from sqlalchemy import select
+from sqlalchemy.orm import sessionmaker
+
+from photo_pipeline.models import Asset, AssetPath, SafetyReview
+
+REPORT_VERSION = 1
+# The donor wrote scores as four-decimal strings; anything outside 0..1 was never
+# something it produced, so it is data corruption rather than a score.
+SCORE_RANGE = (0.0, 1.0)
+# Recorded as the reviewer so an imported score is never mistaken for a judgement.
+REVIEWER = "legacy-csv-import"
+
+
+@dataclass
+class ImportReport:
+    source: str
+    rows: int = 0
+    imported: int = 0
+    skipped_existing: int = 0
+    unchanged: int = 0
+    reviewed: int = 0
+    unmatched: list[str] = field(default_factory=list)
+    rejected: list[dict] = field(default_factory=list)
+    conflicts: list[dict] = field(default_factory=list)
+    generated_at: str = ""
+    schema_version: int = REPORT_VERSION
+
+    def to_dict(self) -> dict:
+        return asdict(self)
+
+    @property
+    def counts(self) -> dict[str, int]:
+        return {
+            "rows": self.rows,
+            "imported": self.imported,
+            "skipped_existing": self.skipped_existing,
+            "unchanged": self.unchanged,
+            "reviewed": self.reviewed,
+            "unmatched": len(self.unmatched),
+            "rejected": len(self.rejected),
+            "conflicts": len(self.conflicts),
+        }
+
+
+class LegacyImportService:
+    def __init__(self, session_factory: sessionmaker) -> None:
+        self._session_factory = session_factory
+
+    def import_nsfw_scores(
+        self, csv_path: Path | str, *, overwrite: bool = False, dry_run: bool = False
+    ) -> ImportReport:
+        """Import ``nsfw_scores.csv`` onto asset identity and report what happened."""
+        path = Path(csv_path)
+        report = ImportReport(source=str(path))
+        if not path.is_file():
+            report.rejected.append({"path": str(path), "reason": "csv_missing"})
+            return self._stamp(report)
+
+        with self._session_factory() as session:
+            # Paths are matched against every occurrence an asset ever had, so a
+            # photo scored before a rename is still recognised.
+            by_path = self._path_index(session)
+            latest = self._latest_reviews(session)
+            seen: Counter[str] = Counter()
+
+            for row in self._rows(path, report):
+                report.rows += 1
+                raw_path, raw_score = row
+                seen[raw_path] += 1
+                if seen[raw_path] > 1:
+                    report.rejected.append({"path": raw_path, "reason": "duplicate_path"})
+                    continue
+                score = _parse_score(raw_score)
+                if score is None:
+                    report.rejected.append(
+                        {"path": raw_path, "reason": "unparsable_score", "value": raw_score}
+                    )
+                    continue
+                asset_id = by_path.get(raw_path)
+                if asset_id is None:
+                    report.unmatched.append(raw_path)
+                    continue
+
+                current = latest.get(asset_id)
+                if current is not None and current.decision:
+                    # A human decided this one; the CSV is older evidence.
+                    report.reviewed += 1
+                    if current.score is None or abs(current.score - score) >= 1e-9:
+                        report.conflicts.append(
+                            {
+                                "path": raw_path,
+                                "asset_id": asset_id,
+                                "current": current.score,
+                                "decision": current.decision,
+                                "csv": score,
+                            }
+                        )
+                    continue
+                if current is not None and current.score is not None:
+                    if abs(current.score - score) < 1e-9:
+                        report.unchanged += 1
+                        continue
+                    report.conflicts.append(
+                        {
+                            "path": raw_path,
+                            "asset_id": asset_id,
+                            "current": current.score,
+                            "decision": None,
+                            "csv": score,
+                        }
+                    )
+                    if not overwrite:
+                        report.skipped_existing += 1
+                        continue
+                session.add(
+                    SafetyReview(
+                        id=str(uuid.uuid4()),
+                        asset_id=asset_id,
+                        score=score,
+                        reviewer=REVIEWER,
+                    )
+                )
+                latest[asset_id] = SafetyReview(asset_id=asset_id, score=score)
+                report.imported += 1
+
+            if dry_run:
+                session.rollback()
+            else:
+                session.commit()
+        return self._stamp(report)
+
+    @staticmethod
+    def _rows(path: Path, report: ImportReport):
+        """Yield ``(path, score)`` pairs, tolerating the donor's own sloppiness."""
+        with open(path, newline="", encoding="utf-8", errors="replace") as handle:
+            for row in csv.DictReader(handle):
+                raw_path = (row.get("path") or "").strip()
+                if not raw_path:
+                    report.rejected.append({"path": "", "reason": "missing_path"})
+                    continue
+                yield raw_path, row.get("nsfw_score")
+
+    @staticmethod
+    def _latest_reviews(session) -> dict[str, SafetyReview]:
+        """The current safety row per asset — latest wins, as everywhere else."""
+        latest: dict[str, SafetyReview] = {}
+        for review in session.scalars(select(SafetyReview).order_by(SafetyReview.created_at)):
+            latest[review.asset_id] = review
+        return latest
+
+    @staticmethod
+    def _path_index(session) -> dict[str, str]:
+        """Every path an asset is or was known by → its stable id."""
+        index: dict[str, str] = {}
+        for asset_id, path in session.execute(select(AssetPath.asset_id, AssetPath.path)):
+            index.setdefault(path, asset_id)
+        for asset_id, path in session.execute(select(Asset.id, Asset.current_path)):
+            if path:
+                index[path] = asset_id  # the current path wins over a closed one
+        return index
+
+    @staticmethod
+    def _stamp(report: ImportReport) -> ImportReport:
+        report.generated_at = datetime.now(timezone.utc).isoformat()
+        return report
+
+
+def write_report(report: ImportReport, directory: Path) -> Path:
+    """Persist the reconciliation report; the import is not evidence until it is."""
+    directory.mkdir(parents=True, exist_ok=True)
+    path = directory / "legacy-nsfw-import.json"
+    path.write_text(json.dumps(report.to_dict(), indent=2, sort_keys=True), encoding="utf-8")
+    return path
+
+
+def _parse_score(value) -> float | None:
+    try:
+        score = float(value)
+    except (TypeError, ValueError):
+        return None
+    if not SCORE_RANGE[0] <= score <= SCORE_RANGE[1]:
+        return None
+    return score

photo_pipeline/services/restores.py

@@ -0,0 +1,650 @@
+"""RestoreService — plan and execute safe restores (US06-04).
+
+Restore is archiving read backwards, with one decisive difference: it removes
+nothing. The archived copy stays on its medium, so every failure mode here costs
+at most a discarded temporary file. What restore must never do is *lose identity*
+— the asset that comes back is the same asset, with its duplicate decision, safety
+review, analysis, and upload history intact — or *overwrite* something in the
+active library.
+
+Preflight proves, per concept §9 "Restore":
+
+- the recorded medium is mounted and is the right one (marker ``media_id``);
+- every selected asset is archived, its archive copy exists, and it hashes to
+  exactly the bytes the database recorded — a mismatch is ``divergent`` and is
+  refused, never silently accepted as "the file";
+- the destination lies inside the library, outside ``_IGNORE/``, and is free; a
+  taken path is answered with a collision-free name, never an overwrite;
+- the library filesystem has room for the scope plus the configured reserve;
+- no rename, archive, or restore lease is holding the lane.
+
+Blocker codes: ``no_library_root``, ``location_offline``, ``wrong_volume``,
+``unsafe_destination``, ``library_not_writable``, ``insufficient_capacity``,
+``lock_conflict``, ``rename_pending``, ``archive_pending``, ``empty_scope``,
+``not_archived``, ``archive_missing``, ``bytes_changed``.
+
+Per item the sequence is:
+
+```
+journal.begin (transferring)   ← intent persisted BEFORE any disk change
+recheck: medium, hash, free destination, asset still archived
+copy to a temporary file beside the destination, fsync, hash it back
+atomically publish into the library
+journal → verified
+current_path = destination, availability = active, path occurrence opened
+journal → complete
+```
+
+Like archiving, the confirmation token is derived from the report, so a changed
+scope, a swapped medium, or a destination that filled up invalidates it.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import os
+import shutil
+import uuid
+from datetime import datetime, timezone
+from pathlib import Path
+
+from sqlalchemy import select
+from sqlalchemy.orm import sessionmaker
+
+from photo_pipeline.config import Config
+from photo_pipeline.jobs.domain_handlers import ARCHIVE_LOCK, LIBRARY_WRITE_LOCK, UPLOAD_LOCK
+from photo_pipeline.models import ArchiveLocation, ArchiveOperation, ArchivePlan, Asset, AssetPath
+from photo_pipeline.path_policy import PathPolicyError, is_excluded, normalize_root, resolve_within
+from photo_pipeline.services import availability
+from photo_pipeline.services.archive_journal import (
+    MANUAL,
+    RESTORE,
+    RESUMABLE,
+    ArchiveJournal,
+    ArchiveState,
+)
+from photo_pipeline.services.archive_transfer import (
+    _clean_temp_files,
+    _fsync_dir,
+    _plan_dict,
+    copy_verify_publish,
+)
+from photo_pipeline.services.archives import ArchiveError
+from photo_pipeline.services.hashing import sha256_file
+from photo_pipeline.services.jobs import JobService
+from photo_pipeline.services.rename_apply import PreconditionFailed, maybe_fault
+from photo_pipeline.services.rename_journal import RenameJournal
+
+PREFLIGHT_VERSION = 1
+TOKEN_PREFIX = f"r{PREFLIGHT_VERSION}"
+# What a restored file is called when its original name is taken. The suffix is
+# visible on purpose: a restore that quietly reuses a name is indistinguishable
+# from an overwrite.
+RESTORED_SUFFIX = "restored"
+
+LOCKS = (LIBRARY_WRITE_LOCK, UPLOAD_LOCK, ARCHIVE_LOCK)
+
+APPLYABLE_PLAN_STATES = frozenset({"planned", "applying", "failed", "complete"})
+
+
+def _now() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+def _issue(code: str, message: str) -> dict:
+    return {"code": code, "message": message}
+
+
+class RestoreService:
+    def __init__(self, session_factory: sessionmaker, *, config: Config) -> None:
+        self._session_factory = session_factory
+        self._config = config
+        self._roots = tuple(normalize_root(root) for root in config.library_roots)
+        self.journal = ArchiveJournal(session_factory)
+
+    # ── preflight ─────────────────────────────────────────────────────────────
+
+    def preflight(self, location_id: str, asset_ids: list[str] | None = None) -> dict:
+        """Validate a restore scope and issue its token. Nothing is written."""
+        with self._session_factory() as session:
+            location = session.get(ArchiveLocation, location_id)
+            if location is None:
+                raise ArchiveError("unknown_location", f"unknown archive location {location_id!r}")
+            root = Path(location.root)
+            online = availability.location_online(location)
+            marker = availability.read_marker(root)
+            report = {
+                "schema_version": PREFLIGHT_VERSION,
+                "location": {
+                    "id": location.id,
+                    "name": location.name,
+                    "root": str(root),
+                    "media_id": location.media_id,
+                    "state": _location_state(root, marker, location.media_id),
+                },
+                "blockers": [],
+            }
+            items = self._items(session, location, asset_ids, reachable=online)
+
+        report["blockers"] += self._destination_blockers(report["location"]["state"], root)
+        report["blockers"] += self._lock_blockers()
+        report["items"] = items
+        report["totals"] = {
+            "assets": len(items),
+            "blocked": sum(1 for item in items if item["blockers"]),
+            "bytes": sum(item["byte_size"] or 0 for item in items),
+        }
+        report["capacity"] = self._capacity(report["totals"]["bytes"])
+        if not report["capacity"]["sufficient"]:
+            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 "
+                    "the library",
+                )
+            )
+        if not items:
+            report["blockers"].append(
+                _issue("empty_scope", "no archived assets are in the selected scope")
+            )
+        report["state"] = (
+            "ready"
+            if not report["blockers"] and not report["totals"]["blocked"]
+            else "blocked"
+        )
+        report["token"] = _token(report)
+        report["generated_at"] = _now().isoformat()
+        return report
+
+    def verify_token(self, token: str, location_id: str, asset_ids: list[str] | None = None) -> bool:
+        return bool(token) and token == self.preflight(location_id, asset_ids)["token"]
+
+    def _items(
+        self, session, location: ArchiveLocation, asset_ids: list[str] | None, *, reachable: bool
+    ) -> list[dict]:
+        stmt = select(Asset).where(Asset.archive_location_id == location.id)
+        if asset_ids is None:
+            # A restored asset keeps its archive link; the default scope is only what
+            # is still archived, so restoring twice is an empty scope, not a blocker.
+            stmt = stmt.where(Asset.availability_state.in_(availability.ARCHIVED))
+        else:
+            stmt = stmt.where(Asset.id.in_(asset_ids))
+        assets = list(session.scalars(stmt.order_by(Asset.archive_path)))
+        if asset_ids is not None:
+            unknown = sorted(set(asset_ids) - {asset.id for asset in assets})
+            if unknown:
+                raise ArchiveError(
+                    "unknown_asset", f"not archived at this location: {', '.join(unknown)}"
+                )
+        taken: set[str] = set()
+        return [self._item(asset, location, reachable=reachable, taken=taken) for asset in assets]
+
+    def _item(self, asset: Asset, location: ArchiveLocation, *, reachable: bool, taken: set) -> dict:
+        source = Path(location.root) / (asset.archive_path or "")
+        blockers: list[dict] = []
+        archive_sha256 = None
+
+        if asset.availability_state not in availability.ARCHIVED:
+            blockers.append(
+                _issue("not_archived", f"asset {asset.id} is {asset.availability_state}")
+            )
+        if reachable:
+            if not source.exists():
+                blockers.append(_issue("archive_missing", f"{source} is not on the medium"))
+            else:
+                archive_sha256 = sha256_file(source)
+                if asset.current_sha256 and archive_sha256 != asset.current_sha256:
+                    blockers.append(
+                        _issue(
+                            "bytes_changed",
+                            f"{source} holds bytes that are not the recorded ones; "
+                            "the archived copy is divergent",
+                        )
+                    )
+
+        destination, destination_blockers = self._destination(asset, taken)
+        blockers += destination_blockers
+        if destination is not None:
+            taken.add(str(destination))
+        return {
+            "asset_id": asset.id,
+            "archive_path": asset.archive_path,
+            "source_path": str(source),
+            "destination_path": str(destination) if destination else None,
+            "expected_sha256": asset.current_sha256,
+            "archive_sha256": archive_sha256,
+            "byte_size": asset.byte_size,
+            "availability_state": asset.availability_state,
+            "blockers": blockers,
+        }
+
+    def _destination(self, asset: Asset, taken: set) -> tuple[Path | None, list[dict]]:
+        """A free path inside the library that mirrors the archived layout.
+
+        Restoring onto an existing file is never an option, so a taken name is
+        answered with ``name (restored).ext`` — visible, ordinary, and impossible to
+        confuse with an overwrite.
+        """
+        if not self._roots:
+            return None, [_issue("no_library_root", "no library root is configured")]
+        root = self._roots[0]
+        try:
+            candidate = resolve_within(root, root / (asset.archive_path or ""))
+        except PathPolicyError as error:
+            return None, [_issue("unsafe_destination", str(error))]
+        if is_excluded(candidate):
+            return None, [
+                _issue("unsafe_destination", f"{candidate} is inside an excluded (_IGNORE/) tree")
+            ]
+        return _free_path(candidate, taken), []
+
+    def _destination_blockers(self, state: str, root: Path) -> list[dict]:
+        blockers: list[dict] = []
+        if not self._roots:
+            blockers.append(_issue("no_library_root", "no library root is configured"))
+        elif not os.access(self._roots[0], os.W_OK):
+            blockers.append(
+                _issue("library_not_writable", f"{self._roots[0]} is not writable")
+            )
+        if state == "offline":
+            blockers.append(
+                _issue("location_offline", f"the archive medium is not mounted at {root}")
+            )
+        elif state == "wrong_volume":
+            blockers.append(_issue("wrong_volume", f"{root} holds a different archive medium"))
+        return blockers
+
+    def _lock_blockers(self) -> list[dict]:
+        blockers: list[dict] = []
+        jobs = JobService(self._session_factory)
+        for lock in LOCKS:
+            held = jobs.blockers(lock)
+            if held:
+                blockers.append(
+                    _issue("lock_conflict", f"the {lock} lane is busy: job {held[0]['id']}")
+                )
+        if RenameJournal(self._session_factory).blocks_mutation():
+            blockers.append(
+                _issue("rename_pending", "an unresolved rename must be recovered before restoring")
+            )
+        if self.journal.blocks_mutation():
+            blockers.append(
+                _issue(
+                    "archive_pending",
+                    "an unresolved archive or restore must be recovered before restoring",
+                )
+            )
+        return blockers
+
+    def _capacity(self, required: int) -> dict:
+        reserve = self._config.archive_free_space_reserve_bytes
+        free = shutil.disk_usage(self._roots[0]).free if self._roots else None
+        return {
+            "required_bytes": required,
+            "reserve_bytes": reserve,
+            "free_bytes": free,
+            "sufficient": free is not None and free >= required + reserve,
+        }
+
+    # ── plans ─────────────────────────────────────────────────────────────────
+
+    def create(self, location_id: str, asset_ids: list[str] | None = None, *, token: str) -> dict:
+        preflight = self.preflight(location_id, asset_ids)
+        if not token or token != preflight["token"]:
+            raise ArchiveError("stale_token", "the restore preflight changed since it was approved")
+        if preflight["state"] != "ready":
+            codes = ", ".join(sorted({issue["code"] for issue in preflight["blockers"]})) or "-"
+            blocked = sorted(
+                {issue["code"] for item in preflight["items"] for issue in item["blockers"]}
+            )
+            raise ArchiveError(
+                "blocked", f"the restore scope is blocked: {', '.join(blocked) or codes}"
+            )
+
+        plan_id = str(uuid.uuid4())
+        with self._session_factory() as session:
+            session.add(
+                ArchivePlan(
+                    id=plan_id,
+                    location_id=location_id,
+                    token=token,
+                    albums=json.dumps(asset_ids) if asset_ids is not None else None,
+                    direction=RESTORE,
+                    state="planned",
+                    schema_version=PREFLIGHT_VERSION,
+                    asset_count=preflight["totals"]["assets"],
+                    byte_size=preflight["totals"]["bytes"],
+                )
+            )
+            session.flush()
+            for sequence, item in enumerate(preflight["items"]):
+                session.add(
+                    ArchiveOperation(
+                        id=str(uuid.uuid4()),
+                        plan_id=plan_id,
+                        direction=RESTORE,
+                        sequence=sequence,
+                        album=Path(item["archive_path"]).parent.name or "(root)",
+                        asset_id=item["asset_id"],
+                        source_path=item["source_path"],
+                        destination_path=item["destination_path"],
+                        archive_path=item["archive_path"],
+                        expected_sha256=item["expected_sha256"],
+                        byte_size=item["byte_size"],
+                        journal_state=ArchiveState.PLANNED,
+                    )
+                )
+            session.commit()
+        return self.get(plan_id)
+
+    def get(self, plan_id: str) -> dict | None:
+        with self._session_factory() as session:
+            plan = session.get(ArchivePlan, plan_id)
+            if plan is None or plan.direction != RESTORE:
+                return None
+            report = _plan_dict(plan)
+        report["operations"] = self.journal.operations(plan_id)
+        return report
+
+    def list(self) -> list[dict]:
+        with self._session_factory() as session:
+            rows = session.scalars(
+                select(ArchivePlan)
+                .where(ArchivePlan.direction == RESTORE)
+                .order_by(ArchivePlan.created_at)
+            )
+            return [_plan_dict(row) for row in rows]
+
+    # ── apply ─────────────────────────────────────────────────────────────────
+
+    def apply(
+        self, plan_id: str, *, expected_version: int | None = None, worker_id: str = "restore"
+    ) -> dict:
+        plan = self._require_plan(plan_id)
+        if expected_version is not None and plan["version"] != expected_version:
+            raise ArchiveError(
+                "stale_plan",
+                f"plan {plan_id} is at version {plan['version']}, expected {expected_version}",
+            )
+        if plan["state"] not in APPLYABLE_PLAN_STATES:
+            raise ArchiveError("invalid_state", f"plan {plan_id} is {plan['state']}")
+        blocking = [row for row in self.journal.incomplete() if row["plan_id"] != plan_id]
+        if blocking:
+            raise ArchiveError(
+                "archive_pending",
+                f"another archive operation is unresolved ({blocking[0]['id']}); recover it first",
+            )
+
+        token = self._claim_plan(plan_id)
+        location = self._location(plan["location_id"])
+        restored = failed = skipped = 0
+        for operation in self.journal.operations(plan_id):
+            if operation["journal_state"] == ArchiveState.COMPLETE:
+                skipped += 1
+                continue
+            try:
+                if operation["journal_state"] == ArchiveState.VERIFIED:
+                    self._finish(operation, token=token)
+                else:
+                    self._restore_one(operation, location, token=token, worker_id=worker_id)
+                restored += 1
+            except PreconditionFailed as error:
+                self._fail(operation, token, error.code, str(error))
+                failed += 1
+            except Exception as error:  # unexpected: record and stop touching disk
+                self._fail(operation, token, "restore_error", str(error))
+                failed += 1
+        state = self.journal.sync_plan_state(plan_id)
+        return {
+            "plan_id": plan_id,
+            "restored": restored,
+            "failed": failed,
+            "skipped": skipped,
+            "state": state,
+        }
+
+    def _restore_one(self, operation: dict, location: dict, *, token: int, worker_id: str) -> None:
+        source = Path(operation["source_path"])
+        destination = Path(operation["destination_path"])
+
+        # 1. Intent first; from here a crash is resolvable from journal + disk.
+        self.journal.begin(operation["id"], worker_id=worker_id, fencing_token=token)
+        maybe_fault(ArchiveState.TRANSFERRING)
+
+        # 2. Recheck against the medium and the library as they are right now.
+        self._recheck(operation, source, destination, location)
+        destination.parent.mkdir(parents=True, exist_ok=True)
+
+        # 3. Always copy: the archived original stays on its medium.
+        copy_verify_publish(source, destination, operation["expected_sha256"])
+        _fsync_dir(destination.parent)
+
+        if sha256_file(destination) != operation["expected_sha256"]:
+            raise PreconditionFailed(
+                "restore_mismatch", f"{destination} does not hold the expected bytes"
+            )
+        self.journal.transition(operation["id"], ArchiveState.VERIFIED, fencing_token=token)
+        maybe_fault(ArchiveState.VERIFIED)
+
+        self._finish(self.journal.get(operation["id"]), token=token)
+
+    def _finish(self, operation: dict, *, token: int) -> None:
+        """Publish the restored file to the database. Idempotent, so recovery may
+        replay it after a crash between the copy and the bookkeeping."""
+        destination = Path(operation["destination_path"])
+        if not destination.exists() or sha256_file(destination) != operation["expected_sha256"]:
+            raise PreconditionFailed(
+                "restore_unverified", f"{destination} is not a verified restored copy"
+            )
+        self._record_restored(operation, destination)
+        self.journal.transition(operation["id"], ArchiveState.COMPLETE, fencing_token=token)
+        maybe_fault(ArchiveState.COMPLETE)
+
+    def _recheck(self, operation: dict, source: Path, destination: Path, location: dict) -> None:
+        root = Path(location["root"])
+        if not root.is_dir() or not (root / availability.MARKER_NAME).exists():
+            raise PreconditionFailed("location_offline", f"{root} is not the archive medium")
+        if not source.exists():
+            raise PreconditionFailed("archive_missing", f"{source} is not on the medium")
+        if source.is_symlink() or destination.is_symlink():
+            raise PreconditionFailed("symlink", "refusing to restore through a symlink")
+        if destination.exists():
+            # Never overwrite: the plan's free path was taken since it was made.
+            raise PreconditionFailed(
+                "destination_exists", f"destination {destination} is occupied"
+            )
+        if not self._inside_library(destination):
+            raise PreconditionFailed(
+                "destination_escape", f"{destination} is outside the library roots"
+            )
+        if sha256_file(source) != operation["expected_sha256"]:
+            self._mark_divergent(operation["asset_id"])
+            raise PreconditionFailed(
+                "bytes_changed", f"{source} changed since the plan was approved"
+            )
+        with self._session_factory() as session:
+            asset = session.get(Asset, operation["asset_id"])
+            if asset is None or asset.availability_state not in availability.ARCHIVED:
+                raise PreconditionFailed(
+                    "not_archived", f"asset {operation['asset_id']} is no longer archived"
+                )
+
+    def _inside_library(self, destination: Path) -> bool:
+        for root in self._roots:
+            try:
+                resolve_within(root, destination)
+                return True
+            except PathPolicyError:
+                continue
+        return False
+
+    # ── database ──────────────────────────────────────────────────────────────
+
+    def _record_restored(self, operation: dict, destination: Path) -> None:
+        """The bytes are back in the library: open the new active occurrence and set
+        availability. Identity, decisions, and history are untouched — that is the
+        entire point of restoring rather than re-importing."""
+        now = _now()
+        with self._session_factory() as session:
+            asset = session.get(Asset, operation["asset_id"])
+            if asset is None:
+                raise PreconditionFailed(
+                    "asset_missing", f"asset {operation['asset_id']} no longer exists"
+                )
+            # A restored asset may be returning to a path it once held, so only an
+            # *open* occurrence counts as already registered — that is what keeps
+            # recovery idempotent without collapsing the path history.
+            recorded = session.scalar(
+                select(AssetPath).where(
+                    AssetPath.asset_id == asset.id,
+                    AssetPath.path == str(destination),
+                    AssetPath.valid_until.is_(None),
+                )
+            )
+            if recorded is None:  # idempotent: recovery may replay this
+                session.add(
+                    AssetPath(
+                        asset_id=asset.id,
+                        path=str(destination),
+                        valid_from=now,
+                        reason="restore",
+                    )
+                )
+            asset.current_path = str(destination)
+            asset.availability_state = availability.ACTIVE
+            asset.missing_at = None
+            # The archive copy stays where it is; keeping the link means a restored
+            # asset still knows which medium holds its archived bytes.
+            asset.archive_divergent_at = None
+            asset.state_version += 1
+            asset.updated_at = now
+            session.commit()
+
+    def _mark_divergent(self, asset_id: str) -> None:
+        """Record that the archived copy is not the recorded file. Durable, because
+        the next restore attempt must not rediscover this from scratch."""
+        with self._session_factory() as session:
+            asset = session.get(Asset, asset_id)
+            if asset is None:
+                return
+            asset.archive_divergent_at = _now()
+            asset.state_version += 1
+            session.commit()
+
+    # ── recovery ──────────────────────────────────────────────────────────────
+
+    def recover(self, *, worker_id: str = "restore-recovery") -> dict:
+        """Resolve every incomplete restore from journal + disk evidence.
+
+        A restore never removed anything, so ``resumable`` simply discards the
+        temporary debris and re-plans the item; ``forward`` finishes the bookkeeping
+        for a published file; ``manual`` is left untouched and keeps blocking.
+        """
+        results = {"resumed": 0, "completed": 0, "manual": 0}
+        touched: set[str] = set()
+        for verdict in self.journal.classify_all(direction=RESTORE):
+            operation = self.journal.get(verdict["operation_id"])
+            touched.add(operation["plan_id"])
+            token = (operation["fencing_token"] or 0) + 1
+            if verdict["classification"] == MANUAL:
+                results["manual"] += 1
+                continue
+            if verdict["classification"] == RESUMABLE:
+                _clean_temp_files(Path(operation["destination_path"]).parent)
+                self.journal.transition(operation["id"], ArchiveState.PLANNED, fencing_token=token)
+                results["resumed"] += 1
+                continue
+            try:
+                self._finish(operation, token=token)
+                results["completed"] += 1
+            except PreconditionFailed as error:
+                self._fail(operation, token, error.code, str(error))
+                results["manual"] += 1
+        for plan_id in touched:
+            self.journal.sync_plan_state(plan_id)
+        return results
+
+    def recovery_status(self) -> dict:
+        verdicts = self.journal.classify_all(direction=RESTORE)
+        return {
+            "operations": verdicts,
+            "manual": [v for v in verdicts if v["classification"] == MANUAL],
+            "blocks_mutation": self.journal.blocks_mutation(),
+        }
+
+    # ── helpers ───────────────────────────────────────────────────────────────
+
+    def _fail(self, operation: dict, token: int, code: str, message: str) -> None:
+        self.journal.transition(
+            operation["id"], ArchiveState.FAILED, fencing_token=token, error=(code, message)
+        )
+
+    def _require_plan(self, plan_id: str) -> dict:
+        plan = self.get(plan_id)
+        if plan is None:
+            raise ArchiveError("unknown_plan", f"unknown restore plan {plan_id!r}")
+        return plan
+
+    def _location(self, location_id: str) -> dict:
+        with self._session_factory() as session:
+            location = session.get(ArchiveLocation, location_id)
+            if location is None:
+                raise ArchiveError("unknown_location", f"unknown archive location {location_id!r}")
+            return {"id": location.id, "root": location.root, "media_id": location.media_id}
+
+    def _claim_plan(self, plan_id: str) -> int:
+        with self._session_factory() as session:
+            plan = session.get(ArchivePlan, plan_id)
+            plan.version += 1
+            plan.state = "applying"
+            plan.updated_at = _now()
+            token = plan.version
+            session.commit()
+        return token
+
+
+# ── module helpers ───────────────────────────────────────────────────────────
+
+
+def _location_state(root: Path, marker: dict | None, media_id: str) -> str:
+    if not root.is_dir() or marker is None:
+        return "offline"
+    return "online" if marker.get("media_id") == media_id else "wrong_volume"
+
+
+def _free_path(candidate: Path, taken: set) -> Path:
+    """``a.jpg`` → ``a (restored).jpg`` → ``a (restored 2).jpg`` …
+
+    ``taken`` holds the destinations already claimed by earlier items of the same
+    plan, so two restores in one scope cannot plan the same path.
+    """
+    if not candidate.exists() and str(candidate) not in taken:
+        return candidate
+    stem, suffix = candidate.stem, candidate.suffix
+    attempt = 1
+    while True:
+        label = RESTORED_SUFFIX if attempt == 1 else f"{RESTORED_SUFFIX} {attempt}"
+        alternative = candidate.with_name(f"{stem} ({label}){suffix}")
+        if not alternative.exists() and str(alternative) not in taken:
+            return alternative
+        attempt += 1
+
+
+def _token(report: dict) -> str:
+    """Digest of everything the report asserts about the scope and the medium.
+
+    Free space is excluded: it drifts constantly without changing what a restore
+    would do, and the capacity verdict itself is part of the digest.
+    """
+    payload = {key: value for key, value in report.items() if key not in ("generated_at", "token")}
+    payload["capacity"] = {
+        key: value for key, value in payload["capacity"].items() if key != "free_bytes"
+    }
+    digest = hashlib.sha256(
+        json.dumps(payload, sort_keys=True, ensure_ascii=False, default=str).encode("utf-8")
+    ).hexdigest()
+    return f"{TOKEN_PREFIX}:{digest}"

photo_pipeline/services/thumbnails.py

@@ -129,7 +129,10 @@ class ThumbnailService:
                 raise ThumbnailUnavailable(f"asset {asset_id} has no readable file")
             source = str(source)
             if source == asset.current_path:
-                self._validate_path(source)  # archive roots lie outside the library
+                # Render the *resolved* path the check approved: revalidating and then
+                # reopening the original name would let a symlink swapped in between
+                # the two steps decide which bytes are served (US07-02).
+                source = str(self._validate_path(source))  # archive roots lie outside
 
         # Rendering happens outside the DB session (no transaction held during I/O).
         try:
@@ -195,20 +198,16 @@ class ThumbnailService:
                 session.commit()
 
     # ── path safety ──────────────────────────────────────────────────────────
-    def _validate_path(self, current_path: str) -> None:
-        path = Path(current_path)
-        if path_policy.is_excluded(path):
-            raise PathNotAllowed(f"excluded path: {current_path}")
-        roots = self._config.library_roots
-        if not roots:
-            return
-        for root in roots:
-            try:
-                path_policy.resolve_within(Path(root), path)
-                return
-            except path_policy.PathPolicyError:
-                continue
-        raise PathNotAllowed(f"path outside configured roots: {current_path}")
+    def _validate_path(self, current_path: str) -> Path:
+        """The resolved path to read, or ``PathNotAllowed``.
+
+        The message names no path: a refusal is returned to the browser, and where
+        the library lives is not the caller's business (US07-02).
+        """
+        try:
+            return path_policy.resolve_in_roots(self._config.library_roots, current_path)
+        except path_policy.PathPolicyError as error:
+            raise PathNotAllowed(str(error)) from None
 
     # ── cache key + rendering ──────────────────────────────────────────────────
     @staticmethod

pyproject.toml

@@ -24,6 +24,8 @@ test = [
 
 [tool.ruff]
 line-length = 100
+# The frozen CLI archive is evidence, not code under maintenance (US07-01).
+extend-exclude = ["legacy_cli_archive"]
 
 [tool.pytest.ini_options]
 testpaths = ["tests"]

tests/characterization/conftest.py

@@ -14,6 +14,11 @@ import pytest
 from PIL import Image
 
 REPO = Path(__file__).resolve().parents[2]
+# The donors are frozen in the read-only archive (US07-01). Only this suite — and
+# the parity test that compares against them — puts that directory on sys.path;
+# production never does, which tests/unit/test_legacy_archive.py enforces.
+ARCHIVED_SOURCES = REPO / "legacy_cli_archive" / "src"
+sys.path.insert(0, str(ARCHIVED_SOURCES))
 sys.path.insert(0, str(REPO))
 
 EXIFTOOL = shutil.which("exiftool")

tests/characterization/test_donor_ledger.py

@@ -1,13 +1,20 @@
-"""Ledger lint (US01-01): every donor-ledger row must carry a real source
-reference, a target location, and either existing characterization test IDs or
-a real pending backlog story."""
+"""Ledger lint (US01-01, extended by US07-01).
+
+Every donor-ledger row must carry a real source reference — now inside the frozen
+archive — a target location, and either existing characterization test IDs or a
+real pending backlog story. Since archival (US07-01) a row may also be ``resolved``:
+its replacement has shipped, ``parity`` names tests that exist and prove it, and
+``delta`` states every intentional difference. Nothing may quietly become
+"finished" without one of those two."""
 import re
 from pathlib import Path
 
 import yaml
 
 REPO = Path(__file__).resolve().parents[2]
-LEDGER = REPO / "donor_ledger.yaml"
+ARCHIVE = REPO / "legacy_cli_archive"
+LEDGER = ARCHIVE / "donor_ledger.yaml"
+ARCHIVED_SOURCES = ARCHIVE / "src"
 STORIES = REPO / "delivery_backlog" / "stories"
 TESTS_DIR = Path(__file__).resolve().parent
 
@@ -15,7 +22,7 @@ CLASSIFICATIONS = {"reuse", "extract", "refactor", "replace"}
 REQUIRED_AREAS = {"discovery", "hashing", "imaging", "nsfw", "vision", "exif",
                   "database", "ui", "configuration", "logging", "cancellation",
                   "error"}
-STATUSES = {"characterized", "pending"}
+STATUSES = {"characterized", "resolved", "pending"}
 
 
 def load_rows():
@@ -49,10 +56,11 @@ def test_rows_have_required_fields_and_unique_ids():
 
 
 def test_source_references_resolve():
+    """Source paths are relative to the archive: the donors moved there, whole."""
     for r in load_rows():
         src = r["source"]
-        f = REPO / src["file"]
-        assert f.is_file(), f"{r['id']}: source file {src['file']} missing"
+        f = ARCHIVED_SOURCES / src["file"]
+        assert f.is_file(), f"{r['id']}: source file {src['file']} missing from the archive"
         text = f.read_text(encoding="utf-8")
         for sym in src["symbols"]:
             assert sym in text, f"{r['id']}: symbol {sym!r} not found in {src['file']}"
@@ -63,7 +71,9 @@ def test_rows_have_tests_or_pending_story():
     for r in load_rows():
         tests = r.get("tests", [])
         pending = r.get("pending_story")
-        assert tests or pending, f"{r['id']}: neither tests nor pending_story"
+        parity = r.get("parity", [])
+        assert tests or pending or parity, \
+            f"{r['id']}: neither characterization tests, parity tests, nor a pending story"
         for t in tests:
             assert t in known_tests, f"{r['id']}: unknown test id {t}"
         if pending:
@@ -73,16 +83,58 @@ def test_rows_have_tests_or_pending_story():
             assert tests, f"{r['id']}: characterized rows need test ids"
 
 
+def test_resolved_rows_name_their_parity_or_their_delta():
+    """A resolved row is a claim that the behavior is handled. It has to say how:
+    tests that prove the replacement, or a stated difference — usually both."""
+    for r in load_rows():
+        if r["status"] != "resolved":
+            assert "parity" not in r, f"{r['id']}: parity on a non-resolved row"
+            continue
+        parity = r.get("parity", [])
+        delta = r.get("delta")
+        assert parity or delta, f"{r['id']}: resolved without parity tests or a delta"
+        if delta:
+            assert len(str(delta).strip()) >= 20, f"{r['id']}: delta too thin to be a reason"
+        for ref in parity:
+            rel, _, func = ref.partition("::")
+            path = REPO / rel
+            assert path.is_file(), f"{r['id']}: parity test file {rel} missing"
+            assert f"def {func}" in path.read_text(encoding="utf-8"), \
+                f"{r['id']}: parity test {ref} not found"
+
+
+def test_pending_rows_are_the_only_unfinished_work():
+    """The ledger is the honest list of what has not been carried over: a pending
+    row names the story that will, and that story must still be open work."""
+    for r in load_rows():
+        if r["status"] != "pending":
+            continue
+        story = r.get("pending_story")
+        assert story, f"{r['id']}: pending without a story"
+        assert list(STORIES.glob(f"{story}-*.md")), f"{r['id']}: unknown story {story}"
+
+
+def test_every_target_module_exists():
+    """A row is only finished if the thing it points at is really there."""
+    for r in load_rows():
+        if r["status"] == "pending":
+            continue
+        modules = re.findall(r"photo_pipeline/[\w/]+\.py", str(r["target"]))
+        for module in modules:
+            assert (REPO / module).is_file(), f"{r['id']}: target {module} does not exist"
+
+
 def test_all_required_areas_covered():
     covered = {r["area"] for r in load_rows()}
     assert REQUIRED_AREAS <= covered, f"uncovered areas: {REQUIRED_AREAS - covered}"
     assert covered <= REQUIRED_AREAS, f"unknown areas: {covered - REQUIRED_AREAS}"
 
 
-def test_no_legacy_file_moved():
-    # US01-01 explicitly forbids moving/archiving donors; the ledger's source
-    # files must all still exist at their original locations.
+def test_every_donor_is_in_the_archive_and_nowhere_else():
+    """US01-01 forbade archiving before characterization; US07-01 requires it after.
+    Each donor exists exactly once — frozen, in the archive."""
     for donor in ("photo_analyzer.py", "nsfwtag/scoring.py", "nsfwtag/exif.py",
                   "nsfwtag/server.py", "webapp/query.py", "webapp/runner.py",
                   "webapp/server.py"):
-        assert (REPO / donor).is_file(), f"donor moved: {donor}"
+        assert (ARCHIVED_SOURCES / donor).is_file(), f"donor missing from archive: {donor}"
+        assert not (REPO / donor).exists(), f"donor still live at the repo root: {donor}"

tests/conftest.py

@@ -1,4 +1,15 @@
-"""Make the repository root importable for the pipeline test suites."""
+"""Make the repository root importable for the pipeline test suites, and give every
+suite the application session the API requires since US07-02.
+
+The suites drive the API the way the browser does — module-level ``httpx`` calls and
+``TestClient`` — so instead of threading a cookie through several hundred call sites,
+both clients bootstrap the session themselves exactly like ``frontend/js/api.js``:
+fetch ``/api/v1/session`` once, then send the cookie plus the CSRF header, and
+re-bootstrap once on 401 (a restarted server issues a new session).
+
+Security tests deliberately bypass this by constructing their own ``httpx.Client``;
+only the module-level helpers are wrapped.
+"""
 
 import sys
 from pathlib import Path
@@ -6,3 +17,85 @@ from pathlib import Path
 REPO = Path(__file__).resolve().parents[1]
 if str(REPO) not in sys.path:
     sys.path.insert(0, str(REPO))
+
+import httpx
+import httpx._api  # httpx.get/post resolve `request` in this module
+import pytest
+from starlette.testclient import TestClient
+
+from photo_pipeline.api.security import CSRF_HEADER, SESSION_COOKIE
+
+_SESSIONS: dict[str, tuple[str, str]] = {}  # base url -> (session id, csrf token)
+
+
+def _base(url) -> str:
+    parts = httpx.URL(str(url))
+    return f"{parts.scheme}://{parts.netloc.decode()}"
+
+
+def _bootstrap(base: str) -> tuple[str, str]:
+    if base not in _SESSIONS:
+        response = httpx.Client(timeout=10).get(f"{base}/api/v1/session")
+        _SESSIONS[base] = (
+            response.cookies.get(SESSION_COOKIE, ""),
+            response.json().get("csrf_token", ""),
+        )
+    return _SESSIONS[base]
+
+
+def session_client(base_url: str, **kwargs) -> httpx.Client:
+    """An ``httpx.Client`` that has bootstrapped its own application session.
+
+    For suites that keep one client for a whole journey; it holds its own cookie, so
+    a client made after a server restart picks up the new session by construction.
+    """
+    client = httpx.Client(base_url=base_url, **kwargs)
+    client.headers[CSRF_HEADER] = client.get("/api/v1/session").json()["csrf_token"]
+    return client
+
+
+def _authorized(url, headers, cookies):
+    session_id, token = _bootstrap(_base(url))
+    headers = dict(headers or {})
+    headers.setdefault(CSRF_HEADER, token)
+    cookies = dict(cookies or {})
+    cookies.setdefault(SESSION_COOKIE, session_id)
+    return headers, cookies
+
+
+@pytest.fixture(autouse=True, scope="session")
+def _api_session():
+    real_request, real_stream = httpx._api.request, httpx._api.stream
+    real_client_request, real_client_init = TestClient.request, TestClient.__init__
+
+    def request(method, url, *, headers=None, cookies=None, **kwargs):
+        sent, jar = _authorized(url, headers, cookies)
+        response = real_request(method, url, headers=sent, cookies=jar, **kwargs)
+        if response.status_code == 401:
+            _SESSIONS.pop(_base(url), None)
+            sent, jar = _authorized(url, headers, cookies)
+            response = real_request(method, url, headers=sent, cookies=jar, **kwargs)
+        return response
+
+    def stream(method, url, *, headers=None, cookies=None, **kwargs):
+        sent, jar = _authorized(url, headers, cookies)
+        return real_stream(method, url, headers=sent, cookies=jar, **kwargs)
+
+    def client_init(self, app, *args, base_url="http://127.0.0.1", **kwargs):
+        # The default "http://testserver" is not a local host, which is exactly what
+        # the Host check refuses; in-process tests are still same-origin callers.
+        real_client_init(self, app, *args, base_url=base_url, **kwargs)
+
+    def client_request(self, method, url, *, headers=None, **kwargs):
+        if CSRF_HEADER not in self.headers:
+            response = real_client_request(self, "GET", "/api/v1/session")
+            self.headers[CSRF_HEADER] = response.json()["csrf_token"]
+        return real_client_request(self, method, url, headers=headers, **kwargs)
+
+    httpx._api.request, httpx.request = request, request
+    httpx._api.stream, httpx.stream = stream, stream
+    TestClient.request, TestClient.__init__ = client_request, client_init
+    yield
+    httpx._api.request, httpx.request = real_request, real_request
+    httpx._api.stream, httpx.stream = real_stream, real_stream
+    TestClient.request, TestClient.__init__ = real_client_request, real_client_init

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_a_pipeline.py

@@ -22,6 +22,8 @@ import numpy as np
 import pytest
 from PIL import Image
 
+from tests.conftest import session_client
+
 REPO = Path(__file__).resolve().parents[2]
 
 
@@ -92,7 +94,7 @@ class ServerController:
                 pytest.fail(f"server exited: {err.decode(errors='replace')}")
             try:
                 if httpx.get(f"{self.base}/api/v1/health/ready", timeout=1).status_code == 200:
-                    self.client = httpx.Client(base_url=self.base, timeout=10)
+                    self.client = session_client(self.base, timeout=10)
                     return
             except httpx.HTTPError:
                 time.sleep(0.2)

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_review_ui.py

@@ -17,6 +17,9 @@ import httpx
 import numpy as np
 import pytest
 from PIL import Image
+from playwright.sync_api import expect
+
+from tests.conftest import session_client
 
 REPO = Path(__file__).resolve().parents[2]
 
@@ -111,7 +114,7 @@ def server(tmp_path):
         proc.terminate()
         pytest.fail("server never became ready")
 
-    client = httpx.Client(base_url=base, timeout=5)
+    client = session_client(base, timeout=5)
 
     def cluster_by_method(method):
         clusters = client.get("/api/v1/duplicates/clusters").json()["items"]
@@ -148,9 +151,9 @@ def test_fuzzy_decision_requires_confirmation(page, server):
     page.get_by_test_id("confirm").wait_for()
     assert "open" in page.get_by_test_id("cluster-state").inner_text()
     page.get_by_test_id("confirm-yes").click()
-    page.wait_for_function(
-        "document.querySelector('[data-testid=cluster-state]').innerText.includes('dismissed')"
-    )
+    # A locator assertion, not wait_for_function: the app's CSP forbids eval, and
+    # Playwright's polling predicate is evaluated as a string in the page (US07-02).
+    expect(page.get_by_test_id("cluster-state")).to_contain_text("dismissed")
 
 
 def test_stale_version_shows_conflict(page, server):
@@ -176,9 +179,9 @@ def test_decision_persists_after_reload(page, server):
     page.goto(f"{server.base}/app/#/duplicates/{cluster['id']}")
     page.get_by_test_id("not-duplicate").click()
     page.get_by_test_id("confirm-yes").click()
-    page.wait_for_function(
-        "document.querySelector('[data-testid=cluster-state]').innerText.includes('dismissed')"
-    )
+    # A locator assertion, not wait_for_function: the app's CSP forbids eval, and
+    # Playwright's polling predicate is evaluated as a string in the page (US07-02).
+    expect(page.get_by_test_id("cluster-state")).to_contain_text("dismissed")
     page.reload()
     page.get_by_test_id("cluster-state").wait_for()
     assert "dismissed" in page.get_by_test_id("cluster-state").inner_text()

tests/e2e/test_security.py

@@ -0,0 +1,280 @@
+"""US07-02 black-box security tests: authorization and path boundaries.
+
+Everything here talks to a real ``photo_pipeline serve`` child process over HTTP with
+its own ``httpx.Client``, deliberately outside the session helper the rest of the
+suite uses — an attacker does not get a bootstrapped client.
+
+The threat is a page in the user's browser, not a remote attacker: the app listens on
+127.0.0.1, so any site the user visits can send requests to it and can point an
+``<img>`` at its media endpoints. Each journey below is one of those attempts.
+"""
+
+from __future__ import annotations
+
+import json
+from contextlib import contextmanager
+from pathlib import Path
+
+import httpx
+import pytest
+
+from tests.e2e._pipeline_harness import Server, image, seed_library
+
+TIMEOUT = 20
+SENTINEL_KEY = "immich-sentinel-9f3a2b"
+SESSION_COOKIE = "pp_session"
+CSRF_HEADER = "X-CSRF-Token"
+
+# A GET, a mutation, and a media endpoint: the three shapes the policy must cover.
+PROTECTED = [
+    ("GET", "/api/v1/workflow", None),
+    ("POST", "/api/v1/albums/proposals", {}),
+    ("GET", "/api/v1/assets/{asset}/thumbnail?size=256", None),
+]
+
+
+@pytest.fixture(scope="module")
+def stack(tmp_path_factory):
+    """One server, one album of two photos, and a credential sentinel in its config."""
+    tmp_path = tmp_path_factory.mktemp("security")
+    seeded = seed_library(tmp_path, {"a": 1, "b": 2}, {"a": "sfw", "b": "sfw"})
+    outside = tmp_path / "outside"
+    outside.mkdir()
+    image(outside / "secret.jpg", 99)
+    server = Server(
+        seeded,
+        extra_env={
+            "PHOTO_PIPELINE_IMMICH_API_KEY": SENTINEL_KEY,
+            "PHOTO_PIPELINE_IMMICH_SERVER_URL": "http://127.0.0.1:1",
+        },
+    ).start()
+    yield server, seeded, outside
+    server.stop()
+
+
+@contextmanager
+def anonymous(server):
+    with httpx.Client(base_url=server.base, timeout=TIMEOUT) as client:
+        yield client
+
+
+@contextmanager
+def authenticated(server):
+    """A browser that has loaded the app: session cookie in the jar, token in a header."""
+    with httpx.Client(base_url=server.base, timeout=TIMEOUT) as client:
+        client.headers[CSRF_HEADER] = client.get("/api/v1/session").json()["csrf_token"]
+        yield client
+
+
+def call(client, method, path, body, asset):
+    return client.request(method, path.format(asset=asset), json=body)
+
+
+def test_the_api_refuses_every_caller_without_a_session(stack):
+    server, seeded, _ = stack
+    with anonymous(server) as client:
+        for method, path, body in PROTECTED:
+            response = call(client, method, path, body, seeded.asset_ids["a"])
+            assert response.status_code == 401, path
+            assert response.json()["error"]["code"] == "unauthenticated"
+        # Liveness and readiness stay open: an orchestrator holds no session.
+        assert client.get("/api/v1/health/ready").status_code == 200
+        assert client.get("/api/v1/health/live").status_code == 200
+
+
+def test_a_guessed_session_cookie_is_refused(stack):
+    server, _, _ = stack
+    with anonymous(server) as client:
+        client.cookies.set(SESSION_COOKIE, "guessed", domain="127.0.0.1")
+        response = client.get("/api/v1/workflow")
+        assert response.status_code == 401
+        assert response.json()["error"]["code"] == "unauthenticated"
+
+
+def test_the_bootstrap_issues_a_strict_httponly_cookie(stack):
+    server, _, _ = stack
+    with anonymous(server) as client:
+        response = client.get("/api/v1/session")
+        cookie = response.headers["set-cookie"].lower()
+        assert "httponly" in cookie and "samesite=strict" in cookie and "path=/" in cookie
+        assert response.json()["csrf_token"]
+        # The token is in the body, which no other origin may read: no CORS header
+        # grants access to it.
+        assert "access-control-allow-origin" not in response.headers
+
+
+def test_a_session_without_the_csrf_token_may_read_but_not_mutate(stack):
+    server, _, _ = stack
+    with authenticated(server) as client:
+        del client.headers[CSRF_HEADER]
+        assert client.get("/api/v1/albums/proposals").status_code == 200
+
+        response = client.post("/api/v1/albums/proposals", json={})
+        assert response.status_code == 403
+        assert response.json()["error"]["code"] == "csrf_failed"
+
+        response = client.post(
+            "/api/v1/albums/proposals", json={}, headers={CSRF_HEADER: "guessed"}
+        )
+        assert response.status_code == 403
+    # And nothing was created behind the refusal.
+    with authenticated(server) as client:
+        assert client.get("/api/v1/albums/proposals").json()["items"] == []
+
+
+def test_a_foreign_origin_cannot_mutate_even_with_a_session(stack):
+    server, _, _ = stack
+    with authenticated(server) as client:
+        for origin in ("http://evil.example", "http://127.0.0.1:1", "null"):
+            response = client.post(
+                "/api/v1/albums/proposals", json={}, headers={"Origin": origin}
+            )
+            assert response.status_code == 403, origin
+            assert response.json()["error"]["code"] == "origin_not_allowed"
+        # This app's own origin is accepted, so the check is not simply refusing all.
+        allowed = client.post(
+            "/api/v1/albums/proposals", json={}, headers={"Origin": server.base}
+        )
+        assert allowed.status_code == 200
+
+
+def test_a_rebinding_host_is_refused(stack):
+    """A name that resolves to 127.0.0.1 makes the browser treat the attacker's page
+    as same-origin. The Host header still carries that name, so it is checked."""
+    server, _, _ = stack
+    with authenticated(server) as client:
+        response = client.get("/api/v1/workflow", headers={"Host": "photos.evil.example"})
+        assert response.status_code == 403
+        assert response.json()["error"]["code"] == "host_not_allowed"
+
+
+def test_media_cannot_be_embedded_by_another_page(stack):
+    server, seeded, _ = stack
+    url = f"/api/v1/assets/{seeded.asset_ids['a']}/thumbnail?size=256"
+    with authenticated(server) as client:
+        # What an <img> on another site produces: no Origin, but a cross-site marker.
+        blocked = client.get(url, headers={"Sec-Fetch-Site": "cross-site"})
+        assert blocked.status_code == 403
+        assert blocked.json()["error"]["code"] == "cross_site_blocked"
+
+        served = client.get(url, headers={"Sec-Fetch-Site": "same-origin"})
+        assert served.status_code == 200
+        assert served.headers["content-type"] == "image/webp"
+        assert served.headers["cross-origin-resource-policy"] == "same-origin"
+        # Photos must never land in a shared cache.
+        assert served.headers["cache-control"].startswith("private")
+
+
+def test_every_response_carries_the_default_headers_and_no_cors(stack):
+    server, seeded, _ = stack
+    with authenticated(server) as client:
+        responses = [
+            client.get("/api/v1/workflow"),
+            client.get(f"/api/v1/assets/{seeded.asset_ids['a']}/thumbnail?size=256"),
+            client.get("/app/"),
+            client.get("/api/v1/does-not-exist"),
+        ]
+    for response in responses:
+        assert response.headers["x-content-type-options"] == "nosniff"
+        assert response.headers["x-frame-options"] == "DENY"
+        assert response.headers["referrer-policy"] == "no-referrer"
+        assert "frame-ancestors 'none'" in response.headers["content-security-policy"]
+        assert "access-control-allow-origin" not in response.headers
+        assert "access-control-allow-credentials" not in response.headers
+
+
+def test_a_traversal_attempt_addresses_nothing(stack):
+    """Identifiers are database keys, not paths: traversal has nowhere to land."""
+    server, _, _ = stack
+    with authenticated(server) as client:
+        for path in (
+            "/api/v1/albums/proposals/..%2F..%2F..%2Fetc%2Fpasswd",
+            "/api/v1/albums/proposals/../../../etc/passwd",
+            "/api/v1/assets/..%2F..%2Fetc%2Fpasswd/thumbnail?size=256",
+            "/api/v1/assets/%2Fetc%2Fpasswd/thumbnail?size=256",
+        ):
+            response = client.get(path)
+            assert response.status_code in (404, 422), path
+            assert "root:" not in response.text, path
+
+
+def test_an_oversized_request_is_refused_before_it_is_parsed(stack):
+    server, _, _ = stack
+    with authenticated(server) as client:
+        response = client.post(
+            "/api/v1/albums/proposals",
+            content=json.dumps({"albums": ["x" * 2_000_000]}),
+            headers={"Content-Type": "application/json"},
+        )
+        assert response.status_code == 413
+        assert response.json()["error"]["code"] == "payload_too_large"
+
+
+def test_a_malformed_request_reports_the_field_and_nothing_else(stack):
+    server, _, _ = stack
+    with authenticated(server) as client:
+        for content in ("{", '{"albums": 5}', ""):
+            response = client.post(
+                "/api/v1/albums/proposals",
+                content=content,
+                headers={"Content-Type": "application/json"},
+            )
+            assert response.status_code == 422, content
+            error = response.json()["error"]
+            assert error["code"] == "invalid_request"
+            assert "Traceback" not in response.text and "photo_pipeline/" not in response.text
+
+
+def test_errors_reveal_neither_the_credential_nor_an_internal(stack):
+    """The API key is configured but must appear nowhere; failures additionally say
+    nothing about where the library lives or how the server is built.
+
+    Successful responses are a different matter: path previews are the point of the
+    rename and archive views, and the operator is the one who owns those paths.
+    """
+    server, seeded, _ = stack
+    with authenticated(server) as client:
+        responses = [
+            client.post("/api/v1/upload-preflight", json={}),
+            client.get("/api/v1/workflow"),
+            client.get("/api/v1/upload-batches"),
+            client.get("/api/v1/assets/unknown-asset/thumbnail?size=256"),
+            client.get("/api/v1/archive-plans/unknown-plan"),
+            client.post("/api/v1/archive-locations", json={"name": "x", "root": "/nope"}),
+            client.get("/api/v1/does-not-exist"),
+        ]
+    for response in responses:
+        assert SENTINEL_KEY not in response.text
+        assert "Traceback" not in response.text
+        if response.status_code >= 400:
+            assert str(seeded.lib) not in response.text
+            assert "photo_pipeline/" not in response.text
+            assert "sqlite" not in response.text.lower()
+
+
+def test_a_symlink_swapped_under_an_asset_cannot_be_served(stack):
+    """TOCTOU on the media path: the file the database points at is replaced by a
+    link to something outside the library between the scan and the request."""
+    server, seeded, outside = stack
+    secret = outside / "secret.jpg"
+    original = seeded.lib / "b.jpg"
+    original.unlink()
+    original.symlink_to(secret)
+
+    with authenticated(server) as client:
+        response = client.get(f"/api/v1/assets/{seeded.asset_ids['b']}/thumbnail?size=256")
+    assert response.status_code == 403
+    assert response.json()["error"]["code"] == "path_not_allowed"
+    # The refusal names no filesystem location, and no bytes escaped with it.
+    assert str(outside) not in response.text and str(seeded.lib) not in response.text
+    assert secret.read_bytes()[:16] not in response.content
+
+
+def test_the_frontend_shell_stays_reachable_without_a_session(stack):
+    """It must load before any JavaScript can ask for a session."""
+    server, _, _ = stack
+    with anonymous(server) as client:
+        response = client.get("/app/", headers={"Sec-Fetch-Site": "none"})
+        assert response.status_code == 200
+        assert response.headers["content-type"].startswith("text/html")
+        assert Path("frontend/index.html").exists()

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/integration/test_legacy_import.py

@@ -0,0 +1,238 @@
+"""Importing the archived CLI's CSV state into the database (US07-01).
+
+The donor's ``nsfw_scores.csv`` was keyed by path; the database is keyed by a
+stable asset id. Every case here is about that gap: a path that moved, a path that
+matches nothing, a score a human has already overruled, and a file that is simply
+malformed. The import may add evidence and must never invent an asset, overwrite a
+judgement, or fail silently — whatever it does ends up in the report.
+"""
+
+from __future__ import annotations
+
+import json
+import uuid
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+import pytest
+from sqlalchemy import select
+
+from photo_pipeline.db import create_db_engine, create_session_factory, run_migrations
+from photo_pipeline.models import Asset, AssetPath, SafetyReview
+from photo_pipeline.services.legacy_import import (
+    REVIEWER,
+    LegacyImportService,
+    write_report,
+)
+
+NOW = datetime(2026, 1, 1, tzinfo=timezone.utc)
+
+
+@pytest.fixture
+def factory(tmp_path):
+    url = f"sqlite:///{tmp_path / 'legacy.db'}"
+    run_migrations(url)
+    engine = create_db_engine(url)
+    yield create_session_factory(engine)
+    engine.dispose()
+
+
+def _asset(sf, path: Path, *, previous: Path | None = None) -> str:
+    asset_id = str(uuid.uuid4())
+    with sf() as session:
+        session.add(
+            Asset(
+                id=asset_id,
+                original_path=str(previous or path),
+                current_path=str(path),
+                discovered_at=NOW,
+                hash_version=1,
+            )
+        )
+        session.add(AssetPath(asset_id=asset_id, path=str(path), valid_from=NOW))
+        if previous is not None:
+            session.add(
+                AssetPath(
+                    asset_id=asset_id, path=str(previous), valid_from=NOW, valid_until=NOW
+                )
+            )
+        session.commit()
+    return asset_id
+
+
+def _review(sf, asset_id: str, *, score=None, decision=None, at=NOW) -> None:
+    with sf() as session:
+        session.add(
+            SafetyReview(
+                id=str(uuid.uuid4()),
+                asset_id=asset_id,
+                score=score,
+                decision=decision,
+                reviewer="dom",
+                created_at=at,
+            )
+        )
+        session.commit()
+
+
+def _csv(tmp_path, rows: str) -> Path:
+    path = tmp_path / "nsfw_scores.csv"
+    path.write_text("path,nsfw_score\n" + rows, encoding="utf-8")
+    return path
+
+
+def _scores(sf) -> dict[str, float]:
+    with sf() as session:
+        return {
+            review.asset_id: review.score
+            for review in session.scalars(
+                select(SafetyReview).order_by(SafetyReview.created_at)
+            )
+        }
+
+
+def test_scores_are_imported_onto_asset_identity(tmp_path, factory):
+    a = _asset(factory, tmp_path / "a.jpg")
+    # This one was scored under its old path and has since moved.
+    b = _asset(factory, tmp_path / "moved" / "b.jpg", previous=tmp_path / "b.jpg")
+    csv_path = _csv(
+        tmp_path, f"{tmp_path / 'a.jpg'},0.9123\n{tmp_path / 'b.jpg'},0.0100\n"
+    )
+
+    report = LegacyImportService(factory).import_nsfw_scores(csv_path)
+
+    assert report.counts == {
+        "rows": 2,
+        "imported": 2,
+        "skipped_existing": 0,
+        "unchanged": 0,
+        "reviewed": 0,
+        "unmatched": 0,
+        "rejected": 0,
+        "conflicts": 0,
+    }
+    assert _scores(factory) == {a: 0.9123, b: 0.0100}
+    # The imported rows are evidence, not judgements: no decision is invented.
+    with factory() as session:
+        rows = list(session.scalars(select(SafetyReview)))
+        assert {r.decision for r in rows} == {None}
+        assert {r.reviewer for r in rows} == {REVIEWER}
+    assert csv_path.exists(), "the CSV is left on disk as its own backup"
+
+
+def test_a_reviewed_asset_is_never_overwritten_by_the_csv(tmp_path, factory):
+    asset_id = _asset(factory, tmp_path / "a.jpg")
+    _review(factory, asset_id, score=0.2, decision="sfw")
+    csv_path = _csv(tmp_path, f"{tmp_path / 'a.jpg'},0.9999\n")
+
+    report = LegacyImportService(factory).import_nsfw_scores(csv_path, overwrite=True)
+
+    assert (report.imported, report.reviewed) == (0, 1)
+    assert report.conflicts[0]["decision"] == "sfw"
+    assert report.conflicts[0]["csv"] == 0.9999
+    with factory() as session:
+        rows = list(session.scalars(select(SafetyReview)))
+        assert len(rows) == 1 and rows[0].decision == "sfw"
+
+
+def test_an_unknown_path_is_reported_never_turned_into_an_asset(tmp_path, factory):
+    _asset(factory, tmp_path / "a.jpg")
+    csv_path = _csv(
+        tmp_path, f"{tmp_path / 'a.jpg'},0.5000\n{tmp_path / 'ghost.jpg'},0.5000\n"
+    )
+
+    report = LegacyImportService(factory).import_nsfw_scores(csv_path)
+
+    assert report.unmatched == [str(tmp_path / "ghost.jpg")]
+    with factory() as session:
+        assert session.scalar(select(Asset).where(Asset.current_path.like("%ghost%"))) is None
+
+
+def test_malformed_rows_are_rejected_with_their_reason(tmp_path, factory):
+    _asset(factory, tmp_path / "a.jpg")
+    csv_path = _csv(
+        tmp_path,
+        f"{tmp_path / 'a.jpg'},0.5000\n"
+        f"{tmp_path / 'a.jpg'},0.5000\n"  # the same path twice
+        f"{tmp_path / 'b.jpg'},not-a-number\n"
+        f"{tmp_path / 'c.jpg'},7.5\n"  # outside 0..1: corruption, not a score
+        ",0.5\n",  # no path at all
+    )
+
+    report = LegacyImportService(factory).import_nsfw_scores(csv_path)
+
+    reasons = sorted(r["reason"] for r in report.rejected)
+    assert reasons == ["duplicate_path", "missing_path", "unparsable_score", "unparsable_score"]
+    assert report.imported == 1
+
+
+def test_rerunning_changes_nothing_and_a_differing_score_needs_overwrite(tmp_path, factory):
+    asset_id = _asset(factory, tmp_path / "a.jpg")
+    csv_path = _csv(tmp_path, f"{tmp_path / 'a.jpg'},0.5000\n")
+    service = LegacyImportService(factory)
+    service.import_nsfw_scores(csv_path)
+
+    again = service.import_nsfw_scores(csv_path)
+    assert (again.imported, again.unchanged) == (0, 1)
+    assert len(_scores(factory)) == 1
+
+    changed = _csv(tmp_path, f"{tmp_path / 'a.jpg'},0.8000\n")
+    refused = service.import_nsfw_scores(changed)
+    assert (refused.imported, refused.skipped_existing) == (0, 1)
+    assert refused.conflicts[0]["current"] == 0.5
+
+    forced = service.import_nsfw_scores(changed, overwrite=True)
+    assert forced.imported == 1
+    with factory() as session:
+        latest = list(session.scalars(select(SafetyReview).order_by(SafetyReview.created_at)))[-1]
+        assert (latest.asset_id, latest.score) == (asset_id, 0.8)
+
+
+def test_a_dry_run_reports_without_writing(tmp_path, factory):
+    _asset(factory, tmp_path / "a.jpg")
+    csv_path = _csv(tmp_path, f"{tmp_path / 'a.jpg'},0.5000\n")
+
+    report = LegacyImportService(factory).import_nsfw_scores(csv_path, dry_run=True)
+
+    assert report.imported == 1
+    assert _scores(factory) == {}, "a dry run must leave the database alone"
+
+
+def test_a_missing_csv_is_a_reported_outcome_not_a_crash(tmp_path, factory):
+    report = LegacyImportService(factory).import_nsfw_scores(tmp_path / "nothing.csv")
+    assert report.rejected == [{"path": str(tmp_path / "nothing.csv"), "reason": "csv_missing"}]
+    assert report.rows == 0
+
+
+def test_the_report_is_written_where_it_can_be_audited(tmp_path, factory):
+    _asset(factory, tmp_path / "a.jpg")
+    csv_path = _csv(tmp_path, f"{tmp_path / 'a.jpg'},0.5000\n")
+    report = LegacyImportService(factory).import_nsfw_scores(csv_path)
+
+    written = write_report(report, tmp_path / "data")
+
+    payload = json.loads(written.read_text(encoding="utf-8"))
+    assert payload["imported"] == 1
+    assert payload["source"] == str(csv_path)
+    assert payload["schema_version"] == 1
+    assert payload["generated_at"]
+
+
+def test_a_later_review_still_wins_after_an_import(tmp_path, factory):
+    """The import is a floor, not a ceiling: a human decision made afterwards is
+    the current state, exactly as everywhere else in the app."""
+    asset_id = _asset(factory, tmp_path / "a.jpg")
+    csv_path = _csv(tmp_path, f"{tmp_path / 'a.jpg'},0.9000\n")
+    LegacyImportService(factory).import_nsfw_scores(csv_path)
+    # The import stamps itself with the wall clock, so "afterwards" has to be too.
+    _review(
+        factory,
+        asset_id,
+        score=0.9,
+        decision="sfw",
+        at=datetime.now(timezone.utc) + timedelta(hours=1),
+    )
+
+    with factory() as session:
+        latest = list(session.scalars(select(SafetyReview).order_by(SafetyReview.created_at)))[-1]
+        assert latest.decision == "sfw"

tests/integration/test_restore.py

@@ -0,0 +1,417 @@
+"""Planning and executing safe restores (US06-04).
+
+Restoring is the one archive operation that can *add* a file to the library, so
+every case here asks two questions: did the right bytes come back under the right
+identity, and did anything already in the library get touched? The media are real
+directories, the hashes are real, and the failure paths assert that the archived
+copy is still exactly where it was — a restore that fails must cost nothing.
+"""
+
+import shutil
+import uuid
+from datetime import datetime, timezone
+
+import numpy as np
+import pytest
+from fastapi.testclient import TestClient
+from PIL import Image
+from sqlalchemy import select
+
+from photo_pipeline.api.app import create_app
+from photo_pipeline.config import Config
+from photo_pipeline.db import create_db_engine, create_session_factory, run_migrations
+from photo_pipeline.models import Asset, AssetPath, SafetyReview, UploadBatch, UploadItem
+from photo_pipeline.services import availability
+from photo_pipeline.services.archive_journal import ArchiveState
+from photo_pipeline.services.archive_transfer import ArchiveTransferService
+from photo_pipeline.services.archives import MARKER_NAME, ArchiveError, ArchiveService
+from photo_pipeline.services.hashing import sha256_file
+from photo_pipeline.services.inventory import InventoryService
+from photo_pipeline.services.restores import RestoreService
+
+pytestmark = pytest.mark.phase_f  # part of the Phase F acceptance gate (US06-06)
+
+NOW = datetime(2026, 1, 1, tzinfo=timezone.utc)
+
+
+# ── environment ──────────────────────────────────────────────────────────────
+
+
+def _env(tmp_path):
+    (tmp_path / "data").mkdir(exist_ok=True)
+    lib = tmp_path / "lib"
+    lib.mkdir(exist_ok=True)
+    archive = tmp_path / "archive"
+    archive.mkdir(exist_ok=True)
+    config = Config.from_env(
+        {
+            "PHOTO_PIPELINE_DATA_DIR": str(tmp_path / "data"),
+            "PHOTO_PIPELINE_LIBRARY_ROOTS": str(lib),
+            "PHOTO_PIPELINE_ARCHIVE_FREE_SPACE_RESERVE_BYTES": "0",
+        }
+    )
+    run_migrations(config.database_url)
+    return config, create_session_factory(create_db_engine(config.database_url)), lib, archive
+
+
+def structured(path, seed, size=(192, 144)):
+    path.parent.mkdir(parents=True, exist_ok=True)
+    rng = np.random.default_rng(seed)
+    w, h = size
+    base = np.zeros((h, w, 3), dtype=np.uint8)
+    for _ in range(5):
+        x0 = int(rng.integers(0, w - 40))
+        y0 = int(rng.integers(0, h - 40))
+        base[y0 : y0 + 40, x0 : x0 + 40] = rng.integers(0, 256, 3)
+    Image.fromarray(base).save(path, quality=95)
+    return path
+
+
+def _archived(sf, config, lib, archive, album="rome", seeds=(1, 2)):
+    """A real album taken all the way through archiving, ready to be restored."""
+    folder = lib / album
+    for index, seed in enumerate(seeds):
+        structured(folder / f"{index}.jpg", seed)
+    scan = InventoryService(sf).scan(lib)
+    with sf() as session:
+        batch_id = str(uuid.uuid4())
+        session.add(
+            UploadBatch(
+                id=batch_id,
+                album=album,
+                folder=str(folder),
+                album_name=album,
+                state="succeeded",
+                preflight_token="v1:test",
+                outcome_state="verified",
+                created_at=NOW,
+            )
+        )
+        for path, asset_id in scan.asset_ids.items():
+            session.add(
+                UploadItem(
+                    batch_id=batch_id,
+                    asset_id=asset_id,
+                    path=path,
+                    sha256=sha256_file(path),
+                    sha1="0" * 40,
+                    state="sent",
+                    outcome="uploaded",
+                )
+            )
+            # A decision that must survive the whole round trip.
+            session.add(
+                SafetyReview(
+                    id=str(uuid.uuid4()),
+                    asset_id=asset_id,
+                    decision="sfw",
+                    score=0.01,
+                    reviewer="test",
+                )
+            )
+        session.commit()
+
+    service = ArchiveService(sf, config=config)
+    location = service.register("external", str(archive))
+    token = service.preflight(location["id"])["token"]
+    transfers = ArchiveTransferService(sf, config=config)
+    plan = transfers.create(location["id"], None, token=token)
+    transfers.apply(plan["id"])
+    return location, scan.asset_ids
+
+
+def _restore(sf, config, location_id, asset_ids=None):
+    service = RestoreService(sf, config=config)
+    token = service.preflight(location_id, asset_ids)["token"]
+    plan = service.create(location_id, asset_ids, token=token)
+    return service, plan, service.apply(plan["id"])
+
+
+def _unmount(archive):
+    (archive / MARKER_NAME).rename(archive / f"{MARKER_NAME}.away")
+
+
+def _assets(sf):
+    with sf() as session:
+        return {asset.id: asset for asset in session.scalars(select(Asset))}
+
+
+def _codes(report):
+    return {issue["code"] for issue in report["blockers"]} | {
+        issue["code"] for item in report["items"] for issue in item["blockers"]
+    }
+
+
+# ── preflight ────────────────────────────────────────────────────────────────
+
+
+def test_preflight_blocks_offline_medium(tmp_path):
+    config, sf, lib, archive = _env(tmp_path)
+    location, _ = _archived(sf, config, lib, archive)
+    _unmount(archive)
+
+    report = RestoreService(sf, config=config).preflight(location["id"])
+    assert report["state"] == "blocked"
+    assert "location_offline" in _codes(report)
+
+
+def test_preflight_blocks_wrong_volume(tmp_path):
+    config, sf, lib, archive = _env(tmp_path)
+    location, _ = _archived(sf, config, lib, archive)
+    (archive / MARKER_NAME).write_text('{"media_id": "someone-elses-disk"}', encoding="utf-8")
+
+    report = RestoreService(sf, config=config).preflight(location["id"])
+    assert report["state"] == "blocked"
+    assert "wrong_volume" in _codes(report)
+
+
+def test_preflight_blocks_changed_archive_bytes(tmp_path):
+    config, sf, lib, archive = _env(tmp_path)
+    location, ids = _archived(sf, config, lib, archive, seeds=(1,))
+    asset_id = next(iter(ids.values()))
+    with sf() as session:
+        archived_file = archive / session.get(Asset, asset_id).archive_path
+    archived_file.write_bytes(b"not the photo that was archived")
+
+    report = RestoreService(sf, config=config).preflight(location["id"])
+    assert report["state"] == "blocked"
+    assert "bytes_changed" in _codes(report)
+    with pytest.raises(ArchiveError) as error:
+        _restore(sf, config, location["id"])
+    assert error.value.code == "blocked"
+
+
+def test_preflight_blocks_insufficient_capacity(tmp_path):
+    config, sf, lib, archive = _env(tmp_path)
+    location, _ = _archived(sf, config, lib, archive)
+    greedy = config.model_copy(
+        update={"archive_free_space_reserve_bytes": 1 << 62}  # more than any disk has
+    )
+
+    report = RestoreService(sf, config=greedy).preflight(location["id"])
+    assert report["state"] == "blocked"
+    assert "insufficient_capacity" in _codes(report)
+
+
+def test_token_changes_with_the_scope(tmp_path):
+    config, sf, lib, archive = _env(tmp_path)
+    location, ids = _archived(sf, config, lib, archive)
+    service = RestoreService(sf, config=config)
+
+    whole = service.preflight(location["id"])["token"]
+    partial = service.preflight(location["id"], [sorted(ids.values())[0]])["token"]
+    assert whole != partial
+    assert service.verify_token(whole, location["id"])
+    assert not service.verify_token(partial, location["id"])
+
+
+# ── restore ──────────────────────────────────────────────────────────────────
+
+
+def test_restore_returns_bytes_identity_and_decisions(tmp_path):
+    config, sf, lib, archive = _env(tmp_path)
+    location, ids = _archived(sf, config, lib, archive)
+    archived_hashes = {
+        asset_id: asset.current_sha256 for asset_id, asset in _assets(sf).items()
+    }
+
+    service, plan, result = _restore(sf, config, location["id"])
+    assert (result["restored"], result["failed"], result["state"]) == (2, 0, "complete")
+
+    for asset_id, asset in _assets(sf).items():
+        assert asset.availability_state == availability.ACTIVE
+        assert asset.current_path == str(lib / asset.archive_path)
+        assert sha256_file(asset.current_path) == archived_hashes[asset_id]
+        # The archived copy is a copy: restoring never empties the medium.
+        assert (archive / asset.archive_path).exists()
+        assert asset.archive_location_id == location["id"]
+
+    with sf() as session:
+        # Identity and decisions survived: same ids, same reviews, new occurrence.
+        assert set(ids.values()) == {a.id for a in session.scalars(select(Asset))}
+        assert {r.decision for r in session.scalars(select(SafetyReview))} == {"sfw"}
+        occurrences = [
+            row.reason
+            for row in session.scalars(
+                select(AssetPath).where(AssetPath.asset_id == sorted(ids.values())[0])
+            )
+        ]
+        assert "restore" in occurrences
+
+
+def test_restore_never_overwrites_a_collision(tmp_path):
+    config, sf, lib, archive = _env(tmp_path)
+    location, ids = _archived(sf, config, lib, archive, seeds=(1,))
+    asset_id = next(iter(ids.values()))
+    with sf() as session:
+        archive_path = session.get(Asset, asset_id).archive_path
+    occupied = lib / archive_path
+    occupied.parent.mkdir(parents=True, exist_ok=True)
+    occupied.write_bytes(b"a different photo already lives here")
+    before = occupied.read_bytes()
+
+    service, plan, result = _restore(sf, config, location["id"])
+    assert result["failed"] == 0
+
+    assert occupied.read_bytes() == before  # untouched
+    restored = _assets(sf)[asset_id].current_path
+    assert restored != str(occupied)
+    assert "(restored)" in restored
+    assert sha256_file(restored) == sha256_file(archive / archive_path)
+
+
+def test_apply_refuses_a_destination_taken_after_planning(tmp_path):
+    config, sf, lib, archive = _env(tmp_path)
+    location, ids = _archived(sf, config, lib, archive, seeds=(1,))
+    service = RestoreService(sf, config=config)
+    token = service.preflight(location["id"])["token"]
+    plan = service.create(location["id"], None, token=token)
+
+    # Someone drops a file exactly where the plan intends to publish.
+    destination = plan["operations"][0]["destination_path"]
+    from pathlib import Path
+
+    Path(destination).parent.mkdir(parents=True, exist_ok=True)
+    Path(destination).write_bytes(b"squatter")
+
+    result = service.apply(plan["id"])
+    assert result["failed"] == 1
+    operation = service.journal.operations(plan["id"])[0]
+    assert operation["journal_state"] == ArchiveState.FAILED
+    assert operation["error_code"] == "destination_exists"
+    assert Path(destination).read_bytes() == b"squatter"
+    assert _assets(sf)[next(iter(ids.values()))].availability_state == availability.ARCHIVED_ONLINE
+
+
+def test_changed_archive_bytes_mark_the_asset_divergent(tmp_path):
+    config, sf, lib, archive = _env(tmp_path)
+    location, ids = _archived(sf, config, lib, archive, seeds=(1,))
+    asset_id = next(iter(ids.values()))
+    service = RestoreService(sf, config=config)
+    token = service.preflight(location["id"])["token"]
+    plan = service.create(location["id"], None, token=token)
+
+    # The medium's copy is edited after the plan was approved.
+    with sf() as session:
+        archived_file = archive / session.get(Asset, asset_id).archive_path
+    archived_file.write_bytes(b"edited on the shelf")
+
+    result = service.apply(plan["id"])
+    assert result["failed"] == 1
+    operation = service.journal.operations(plan["id"])[0]
+    assert operation["error_code"] == "bytes_changed"
+    asset = _assets(sf)[asset_id]
+    assert asset.archive_divergent_at is not None  # durable divergence
+    assert asset.availability_state == availability.ARCHIVED_ONLINE
+    assert asset.current_path is None  # nothing was published
+
+
+# ── interruption and idempotency ─────────────────────────────────────────────
+
+
+def test_interrupted_before_publishing_is_resumable(tmp_path):
+    config, sf, lib, archive = _env(tmp_path)
+    location, ids = _archived(sf, config, lib, archive, seeds=(1,))
+    service = RestoreService(sf, config=config)
+    token = service.preflight(location["id"])["token"]
+    plan = service.create(location["id"], None, token=token)
+    operation = service.journal.operations(plan["id"])[0]
+
+    # Model a kill right after the intent was written: nothing published yet.
+    service.journal.begin(operation["id"], worker_id="killed", fencing_token=1)
+
+    status = service.recovery_status()
+    assert status["operations"][0]["classification"] == "resumable"
+    assert service.recover() == {"resumed": 1, "completed": 0, "manual": 0}
+    assert service.journal.operations(plan["id"])[0]["journal_state"] == ArchiveState.PLANNED
+
+    result = service.apply(plan["id"])
+    assert result["failed"] == 0
+    assert _assets(sf)[next(iter(ids.values()))].availability_state == availability.ACTIVE
+
+
+def test_interrupted_after_publishing_is_finished_by_recovery(tmp_path):
+    config, sf, lib, archive = _env(tmp_path)
+    location, ids = _archived(sf, config, lib, archive, seeds=(1,))
+    asset_id = next(iter(ids.values()))
+    service = RestoreService(sf, config=config)
+    token = service.preflight(location["id"])["token"]
+    plan = service.create(location["id"], None, token=token)
+    operation = service.journal.operations(plan["id"])[0]
+
+    # Model a kill between the published copy and the database update.
+    from pathlib import Path
+
+    destination = Path(operation["destination_path"])
+    destination.parent.mkdir(parents=True, exist_ok=True)
+    shutil.copy2(operation["source_path"], destination)
+    service.journal.begin(operation["id"], worker_id="killed", fencing_token=1)
+    service.journal.transition(operation["id"], ArchiveState.VERIFIED, fencing_token=1)
+
+    assert service.recovery_status()["operations"][0]["classification"] == "forward"
+    assert service.recover()["completed"] == 1
+    asset = _assets(sf)[asset_id]
+    assert asset.availability_state == availability.ACTIVE
+    assert asset.current_path == str(destination)
+
+    # Repeated recovery and a repeated apply converge on the same state.
+    assert service.recover() == {"resumed": 0, "completed": 0, "manual": 0}
+    again = service.apply(plan["id"])
+    assert (again["skipped"], again["failed"]) == (1, 0)
+    assert _assets(sf)[asset_id].current_path == str(destination)
+
+
+def test_restored_state_survives_restart_and_rescan(tmp_path):
+    config, sf, lib, archive = _env(tmp_path)
+    location, ids = _archived(sf, config, lib, archive)
+    _restore(sf, config, location["id"])
+
+    restarted = create_session_factory(create_db_engine(config.database_url))
+    InventoryService(restarted).scan(lib)
+
+    assets = _assets(restarted)
+    assert set(assets) == set(ids.values())  # no new identities from the rescan
+    for asset in assets.values():
+        assert asset.availability_state == availability.ACTIVE
+        assert asset.missing_at is None
+
+    # Nothing is archived at that location any more, so there is nothing to restore.
+    again = RestoreService(restarted, config=config).preflight(location["id"])
+    assert _codes(again) == {"empty_scope"}
+
+
+def test_restore_api_round_trip(tmp_path):
+    config, sf, lib, archive = _env(tmp_path)
+    location, ids = _archived(sf, config, lib, archive, seeds=(1,))
+
+    with TestClient(create_app(config)) as client:
+        report = client.post(
+            "/api/v1/restore-preflight", json={"location_id": location["id"]}
+        ).json()
+        assert report["state"] == "ready"
+
+        stale = client.post(
+            "/api/v1/restore-plans",
+            json={"location_id": location["id"], "token": "r1:not-the-token"},
+        )
+        assert stale.status_code == 409
+
+        created = client.post(
+            "/api/v1/restore-plans",
+            json={"location_id": location["id"], "token": report["token"]},
+        )
+        assert created.status_code == 201
+        plan_id = created.json()["id"]
+        assert created.json()["direction"] == "restore"
+
+        # The plan is visible and applying it queues work on the archiver lane.
+        assert client.get(f"/api/v1/restore-plans/{plan_id}").status_code == 200
+        queued = client.post(f"/api/v1/restore-plans/{plan_id}/apply")
+        assert queued.status_code == 200
+        assert queued.json()["job"]["job_type"] == "restore_plan"
+        assert queued.json()["job"]["lock_key"] == "archive"
+        assert client.get("/api/v1/restore-recovery").json()["manual"] == []
+
+    assert _assets(sf)[next(iter(ids.values()))].availability_state == (
+        availability.ARCHIVED_ONLINE  # the worker, not the request, does the work
+    )

tests/integration/test_safety_parity.py

@@ -3,6 +3,8 @@ same fixture files (real exiftool-written keywords)."""
 
 import shutil
 import subprocess
+import sys
+from pathlib import Path
 
 import numpy as np
 import pytest
@@ -11,6 +13,10 @@ from PIL import Image
 from photo_pipeline.integrations import exiftool
 from photo_pipeline.services import safety
 
+# The donor is frozen in the read-only archive (US07-01); parity still compares
+# against it, so this suite is the second and last place that may import it.
+ARCHIVED_SOURCES = Path(__file__).resolve().parents[2] / "legacy_cli_archive" / "src"
+
 EXIFTOOL = shutil.which("exiftool")
 pytestmark = pytest.mark.skipif(EXIFTOOL is None, reason="exiftool not installed")
 
@@ -31,6 +37,7 @@ def _tag(path, keyword):
 
 
 def test_extracted_marks_and_partition_match_donor(tmp_path):
+    sys.path.insert(0, str(ARCHIVED_SOURCES))  # the donor lives in the archive now
     donor_exif = pytest.importorskip("nsfwtag.exif")
 
     nsfw = _jpeg(tmp_path / "nsfw.jpg", 1)

tests/integration/test_security_boundaries.py

@@ -0,0 +1,136 @@
+"""US07-02: what an unhandled failure says, and where the library ends.
+
+Two boundaries that only show up below the HTTP surface:
+
+* Exception text is where internals leak — absolute paths, SQL, and occasionally a
+  credential passed to the call that blew up. The first tests drive the real
+  application with a route that raises such an exception (no production route does)
+  and assert the client sees only a code.
+* The path the database recorded is not the path the filesystem will open a moment
+  later. Analysis is the one stage whose bytes leave this machine, so it resolves
+  the source against the library roots immediately before the provider call.
+"""
+
+from __future__ import annotations
+
+import logging
+import uuid
+from datetime import datetime, timezone
+
+import pytest
+from PIL import Image
+from starlette.testclient import TestClient
+
+from photo_pipeline.api.app import create_app
+from photo_pipeline.config import Config
+from photo_pipeline.db import create_db_engine, create_session_factory, run_migrations
+from photo_pipeline.models import AnalysisResult, Asset
+from photo_pipeline.services.analysis import AnalysisService
+from photo_pipeline.services.safety import SafetyService
+
+BOOM = "sqlite:///Users/someone/Pictures/private.db failed with key sk-secret-123"
+
+
+@pytest.fixture
+def client(tmp_path):
+    app = create_app(Config(data_dir=tmp_path / "data"))
+
+    @app.get("/api/v1/boom")
+    def boom():
+        raise RuntimeError(BOOM)
+
+    with TestClient(app, raise_server_exceptions=False) as test_client:
+        yield test_client
+
+
+def test_an_unhandled_error_returns_a_bare_envelope(client, caplog):
+    with caplog.at_level(logging.ERROR):
+        response = client.get("/api/v1/boom")
+
+    assert response.status_code == 500
+    assert response.json() == {"error": {"code": "internal_error", "message": "internal error"}}
+    assert BOOM not in response.text and "Traceback" not in response.text
+    # The operator still gets the whole story, on the server side.
+    assert BOOM in caplog.text
+
+
+def test_a_refusal_response_still_carries_the_default_headers(client):
+    """A 500 escaping the middleware's response path would also escape its headers."""
+    response = client.get("/api/v1/boom")
+    assert response.headers["x-content-type-options"] == "nosniff"
+    assert response.headers["x-frame-options"] == "DENY"
+
+
+# ── the library boundary, revalidated at the moment of use ───────────────────
+
+
+class RecordingProvider:
+    def __init__(self):
+        self.calls = []
+
+    def analyze(self, path, *, album_hint):
+        self.calls.append(path)
+        return {"description": "a photo", "tags": []}
+
+
+def _library(tmp_path):
+    lib = tmp_path / "lib"
+    lib.mkdir()
+    outside = tmp_path / "outside"
+    outside.mkdir()
+    for path in (lib / "inside.jpg", outside / "private.jpg"):
+        Image.new("RGB", (8, 8), "blue").save(path)
+    (tmp_path / "data").mkdir()
+    config = Config.from_env({"PHOTO_PIPELINE_DATA_DIR": str(tmp_path / "data")})
+    run_migrations(config.database_url)
+    return lib, outside, create_session_factory(create_db_engine(config.database_url))
+
+
+def _sfw_asset(sf, path):
+    asset = Asset(
+        id=str(uuid.uuid4()),
+        original_path=str(path),
+        current_path=str(path),
+        discovered_at=datetime.now(timezone.utc),
+        hash_version=1,
+    )
+    with sf() as session:
+        session.add(asset)
+        session.commit()
+    SafetyService(sf).decide(asset.id, "sfw", write_exif=False)
+    return asset.id
+
+
+def test_analysis_will_not_send_a_file_that_left_the_library(tmp_path):
+    """A link swapped under an asset after the scan points at something the user
+    never put in the library. Those bytes must not reach the vision provider — it is
+    the one place in the pipeline where content leaves this machine."""
+    lib, outside, sf = _library(tmp_path)
+    inside = lib / "inside.jpg"
+    asset_id = _sfw_asset(sf, inside)
+    inside.unlink()
+    inside.symlink_to(outside / "private.jpg")
+
+    provider = RecordingProvider()
+    result = AnalysisService(sf, provider=provider, library_roots=(lib,)).run([asset_id])
+
+    assert provider.calls == [], "the provider must never have been constructed a request"
+    assert result == {"analyzed": 0, "skipped": 0, "errors": 1}
+    with sf() as session:
+        row = session.get(AnalysisResult, asset_id)
+    # The failure is visible and names no path.
+    assert row.status == "error"
+    assert "outside the configured library roots" in row.error_message
+    assert str(outside) not in row.error_message
+
+
+def test_analysis_still_reads_a_file_that_stayed_inside(tmp_path):
+    """The guard must resolve real paths, not refuse everything."""
+    lib, _, sf = _library(tmp_path)
+    asset_id = _sfw_asset(sf, lib / "inside.jpg")
+
+    provider = RecordingProvider()
+    result = AnalysisService(sf, provider=provider, library_roots=(lib,)).run([asset_id])
+
+    assert result["analyzed"] == 1
+    assert provider.calls == [str((lib / "inside.jpg").resolve())]

tests/story_traceability.json

@@ -134,6 +134,24 @@
     ],
     "US06-03": [
       "tests/integration/test_offline_assets.py"
+    ],
+    "US06-04": [
+      "tests/integration/test_restore.py"
+    ],
+    "US06-05": [
+      "tests/e2e/test_archive_ui.py"
+    ],
+    "US06-06": [
+      "tests/e2e/test_phase_f_pipeline.py"
+    ],
+    "US07-01": [
+      "tests/unit/test_legacy_archive.py",
+      "tests/integration/test_legacy_import.py"
+    ],
+    "US07-02": [
+      "tests/unit/test_security_policy.py",
+      "tests/integration/test_security_boundaries.py",
+      "tests/e2e/test_security.py"
     ]
   }
 }

tests/unit/test_legacy_archive.py

@@ -0,0 +1,173 @@
+"""Archive lint (US07-01): the frozen CLI archive is complete, honest, and inert.
+
+Freezing the donors is only worth something if three things stay true: the archive
+still holds exactly the bytes it claims, it holds no secret, and production cannot
+reach it. Each of those is a one-line mistake away — a helpful `sys.path` insert, a
+copied `.env`, an edited "just this once" source — so each is asserted here.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import re
+import subprocess
+import sys
+from pathlib import Path
+
+import pytest
+import yaml
+
+REPO = Path(__file__).resolve().parents[2]
+ARCHIVE = REPO / "legacy_cli_archive"
+SOURCES = ARCHIVE / "src"
+LEDGER = ARCHIVE / "donor_ledger.yaml"
+CHECKSUMS = ARCHIVE / "CHECKSUMS.sha256"
+
+# Everything the concept requires an archive to carry (§3 "Donor-first CLI
+# migration and archival"): the sources, their docs, a dependency lock, schema
+# notes, a redacted sample configuration, the ledger, and recorded checksums.
+REQUIRED_ARTIFACTS = (
+    "README.md",
+    "donor_ledger.yaml",
+    "CHECKSUMS.sha256",
+    "requirements-lock.txt",
+    "photo_analyzer.env.sample",
+)
+ARCHIVED_MODULES = ("photo_analyzer", "nsfwtag", "webapp", "nsfw_tag", "compare_models")
+# Only the two suites that compare against the donors may put the archive on the
+# import path; every other tree must not name it at all.
+IMPORT_ALLOWED = {
+    REPO / "tests" / "characterization" / "conftest.py",
+    REPO / "tests" / "characterization" / "test_donor_ledger.py",  # lints the ledger there
+    REPO / "tests" / "integration" / "test_safety_parity.py",
+    Path(__file__),
+}
+SECRET_PATTERNS = (
+    re.compile(r"sk-(?!REPLACE_WITH_YOUR_KEY)[A-Za-z0-9_\-]{16,}"),
+    re.compile(r"AIza[0-9A-Za-z_\-]{20,}"),  # Google API keys
+    re.compile(r"(?i)api[_-]?key\s*[=:]\s*['\"][A-Za-z0-9_\-]{16,}['\"]"),
+)
+
+
+def _archived_files() -> list[Path]:
+    return sorted(p for p in SOURCES.rglob("*") if p.is_file() and "__pycache__" not in p.parts)
+
+
+# ── completeness ─────────────────────────────────────────────────────────────
+
+
+def test_the_archive_carries_every_required_artifact():
+    for artifact in REQUIRED_ARTIFACTS:
+        path = ARCHIVE / artifact
+        assert path.is_file(), f"archive is missing {artifact}"
+        assert path.stat().st_size > 0, f"archive artifact {artifact} is empty"
+    assert (SOURCES / "photo_analyzer.py").is_file()
+    assert (SOURCES / "nsfwtag" / "README.md").is_file(), "donor docs must be archived too"
+    assert (SOURCES / "webapp" / "README.md").is_file()
+
+
+def test_every_ledger_source_is_present_in_the_archive():
+    rows = yaml.safe_load(LEDGER.read_text(encoding="utf-8"))["rows"]
+    for row in rows:
+        source = SOURCES / row["source"]["file"]
+        assert source.is_file(), f"{row['id']}: {row['source']['file']} is not archived"
+
+
+def test_the_readme_records_provenance_and_the_no_import_rule():
+    readme = (ARCHIVE / "README.md").read_text(encoding="utf-8")
+    for expected in ("CHECKSUMS.sha256", "requirements-lock.txt", "donor_ledger.yaml"):
+        assert expected in readme, f"README does not point at {expected}"
+    assert "Schema notes" in readme, "the donor's schema must be documented"
+    assert "nsfw_scores.csv" in readme, "the CSV's fate must be documented"
+
+
+# ── integrity ────────────────────────────────────────────────────────────────
+
+
+def test_every_archived_source_matches_its_checksum():
+    """A frozen archive that silently drifts is not evidence of anything."""
+    recorded = {}
+    for line in CHECKSUMS.read_text(encoding="utf-8").splitlines():
+        digest, _, rel = line.partition("  ")
+        if rel:
+            recorded[rel.strip()] = digest
+    actual = {
+        str(path.relative_to(ARCHIVE)): hashlib.sha256(path.read_bytes()).hexdigest()
+        for path in _archived_files()
+    }
+    assert actual == recorded, "archived sources and CHECKSUMS.sha256 disagree"
+
+
+# ── redaction ────────────────────────────────────────────────────────────────
+
+
+def test_the_archive_contains_no_credential():
+    for path in [*_archived_files(), *(ARCHIVE / a for a in REQUIRED_ARTIFACTS)]:
+        if path.suffix in (".png", ".jpg", ".webp"):
+            continue
+        text = path.read_text(encoding="utf-8", errors="replace")
+        for pattern in SECRET_PATTERNS:
+            assert not pattern.search(text), f"possible secret in {path.relative_to(REPO)}"
+
+
+def test_the_sample_configuration_is_a_placeholder_only():
+    sample = (ARCHIVE / "photo_analyzer.env.sample").read_text(encoding="utf-8")
+    assert "REDACTED" in sample
+    # Placeholders are not key-shaped, so neither a scanner nor a reader can
+    # mistake the sample for a credential.
+    assert "LLM_API_KEY=<" in sample
+    # No real env file, database, log, or CSV may ride along in the archive.
+    strays = [
+        p.name
+        for p in ARCHIVE.rglob("*")
+        if p.is_file() and p.suffix in (".env", ".db", ".sqlite", ".sqlite3", ".log", ".csv")
+    ]
+    assert strays == [], f"unexpected runtime files archived: {strays}"
+
+
+# ── inertness ────────────────────────────────────────────────────────────────
+
+
+def test_production_code_never_imports_an_archived_module():
+    for path in (REPO / "photo_pipeline").rglob("*.py"):
+        text = path.read_text(encoding="utf-8")
+        for module in ARCHIVED_MODULES:
+            assert not re.search(rf"^\s*(import|from)\s+{module}\b", text, re.MULTILINE), \
+                f"{path.relative_to(REPO)} imports the archived {module}"
+        assert "legacy_cli_archive" not in text, \
+            f"{path.relative_to(REPO)} names the archive"
+
+
+def test_only_the_parity_suites_put_the_archive_on_the_import_path():
+    for path in (REPO / "tests").rglob("*.py"):
+        if path in IMPORT_ALLOWED or "__pycache__" in path.parts:
+            continue
+        assert "legacy_cli_archive" not in path.read_text(encoding="utf-8"), \
+            f"{path.relative_to(REPO)} reaches into the archive"
+
+
+def test_the_archived_modules_are_unimportable_from_a_clean_interpreter():
+    """The real check: a fresh process with the repo on its path cannot load them."""
+    script = (
+        "import importlib.util, sys; "
+        f"sys.path.insert(0, {str(REPO)!r}); "
+        "print([m for m in "
+        f"{list(ARCHIVED_MODULES)!r}"
+        " if importlib.util.find_spec(m) is not None])"
+    )
+    result = subprocess.run(
+        [sys.executable, "-c", script], capture_output=True, text=True, cwd=str(REPO), check=True
+    )
+    assert result.stdout.strip() == "[]", f"still importable: {result.stdout.strip()}"
+
+
+def test_the_application_starts_without_the_archive(tmp_path):
+    """Nothing in the runtime path may need the frozen sources to exist."""
+    pytest.importorskip("fastapi")
+    from photo_pipeline.api.app import create_app
+    from photo_pipeline.config import Config
+
+    config = Config.from_env(
+        {"PHOTO_PIPELINE_DATA_DIR": str(tmp_path / "data"), "PHOTO_PIPELINE_LIBRARY_ROOTS": ""}
+    )
+    assert create_app(config) is not None

tests/unit/test_path_policy.py

@@ -81,3 +81,42 @@ def test_symlink_within_root_is_allowed(tmp_path):
         pytest.skip("cannot create symlink on this platform")
     found = path_policy.discover([root])
     assert target in found and link in found
+
+
+def test_resolve_in_roots_returns_the_path_the_caller_must_use(tmp_path):
+    """The resolved path is the answer, not a yes/no: a caller that revalidates one
+    name and then opens another has an open symlink race (US07-02)."""
+    root = tmp_path / "lib"
+    (root / "sub").mkdir(parents=True)
+    target = root / "sub" / "real.jpg"
+    target.write_bytes(b"x")
+    link = root / "alias.jpg"
+    os.symlink(target, link)
+
+    assert path_policy.resolve_in_roots([root], link) == target.resolve()
+    assert path_policy.resolve_in_roots([root], target) == target.resolve()
+
+
+def test_resolve_in_roots_refuses_escapes_without_naming_them(tmp_path):
+    root = tmp_path / "lib"
+    root.mkdir()
+    outside = tmp_path / "outside.jpg"
+    outside.write_bytes(b"x")
+    link = root / "alias.jpg"
+    os.symlink(outside, link)
+
+    for candidate in (link, outside, root / "_IGNORE" / "a.jpg"):
+        with pytest.raises(path_policy.PathPolicyError) as raised:
+            path_policy.resolve_in_roots([root], candidate)
+        assert str(outside) not in str(raised.value)
+
+
+def test_resolve_in_roots_checks_every_configured_root(tmp_path):
+    first, second = tmp_path / "one", tmp_path / "two"
+    first.mkdir()
+    second.mkdir()
+    photo = second / "b.jpg"
+    photo.write_bytes(b"x")
+    assert path_policy.resolve_in_roots([first, second], photo) == photo.resolve()
+    # No configured boundary means nothing to check against.
+    assert path_policy.resolve_in_roots([], photo) == photo

tests/unit/test_security_policy.py

@@ -0,0 +1,169 @@
+"""US07-02: the request-admission policy, enumerated.
+
+``evaluate`` decides every refusal the API can make before a route runs, so the
+whole local-web threat model is one table here: who may call, from where, with what
+proof. The middleware and the endpoints are covered black box in
+``tests/e2e/test_security.py``; this file pins the rules themselves, including the
+combinations a browser can produce but a test client rarely does.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from photo_pipeline.api.security import (
+    CSRF_HEADER,
+    PUBLIC_PATHS,
+    Session,
+    evaluate,
+    split_host,
+)
+
+SESSION = Session(id="session-id", csrf_token="csrf-token")
+HOST = "127.0.0.1:8000"
+LIMIT = 1024
+
+
+def check(method="GET", path="/api/v1/workflow", **headers):
+    """Evaluate a request that is authenticated and same-origin unless overridden."""
+    sent = {
+        "host": HOST,
+        "cookie-session": SESSION.id,
+        CSRF_HEADER: SESSION.csrf_token,
+    }
+    sent.update({name.replace("_", "-"): value for name, value in headers.items()})
+    sent = {name: value for name, value in sent.items() if value is not None}
+    return evaluate(
+        method=method,
+        path=path,
+        headers=sent,
+        session=SESSION,
+        max_request_bytes=LIMIT,
+    )
+
+
+def test_an_authenticated_same_origin_request_is_admitted():
+    assert check() is None
+    assert check(method="POST", origin="http://127.0.0.1:8000") is None
+    assert check(sec_fetch_site="same-origin") is None
+
+
+@pytest.mark.parametrize("host", ["evil.example", "evil.example:8000", "192.168.1.10:8000", ""])
+def test_a_non_loopback_host_is_refused(host):
+    """DNS rebinding: the browser thinks it is talking to the attacker's name, which
+    resolves to 127.0.0.1. The name is the evidence, so the name is checked."""
+    refusal = check(host=host)
+    assert (refusal.status, refusal.code) == (403, "host_not_allowed")
+
+
+@pytest.mark.parametrize("host", ["127.0.0.1:8000", "localhost:8000", "[::1]:8000", "localhost"])
+def test_loopback_hosts_are_accepted(host):
+    assert check(host=host) is None
+
+
+@pytest.mark.parametrize(
+    "origin",
+    [
+        "http://evil.example",
+        "https://evil.example:8000",
+        "http://127.0.0.1:9999",  # another local app is still another origin
+        "http://localhost.evil.example:8000",
+        "null",
+        "file://",
+    ],
+)
+def test_a_foreign_origin_is_refused(origin):
+    refusal = check(method="POST", origin=origin)
+    assert (refusal.status, refusal.code) == (403, "origin_not_allowed")
+
+
+@pytest.mark.parametrize("origin", ["http://127.0.0.1:8000", "http://localhost:8000"])
+def test_this_applications_origin_is_accepted(origin):
+    assert check(method="POST", origin=origin) is None
+
+
+@pytest.mark.parametrize("site", ["cross-site", "same-site"])
+def test_a_cross_site_fetch_is_refused_even_without_an_origin(site):
+    """What ``<img src="http://127.0.0.1:8000/...">`` on another page looks like."""
+    refusal = check(path="/api/v1/assets/a1/thumbnail", sec_fetch_site=site)
+    assert (refusal.status, refusal.code) == (403, "cross_site_blocked")
+
+
+def test_a_user_initiated_navigation_is_accepted():
+    assert check(sec_fetch_site="none") is None
+
+
+def test_a_request_without_a_session_is_unauthenticated():
+    for method, path in [
+        ("GET", "/api/v1/workflow"),
+        ("POST", "/api/v1/jobs"),
+        ("GET", "/api/v1/assets/a1/thumbnail"),
+    ]:
+        refusal = check(method=method, path=path, cookie_session=None)
+        assert (refusal.status, refusal.code) == (401, "unauthenticated"), path
+
+
+def test_a_forged_session_is_unauthenticated():
+    refusal = check(cookie_session="guessed")
+    assert (refusal.status, refusal.code) == (401, "unauthenticated")
+
+
+@pytest.mark.parametrize("path", sorted(PUBLIC_PATHS))
+def test_health_and_the_bootstrap_stay_reachable_without_a_session(path):
+    assert check(path=path, cookie_session=None) is None
+
+
+def test_the_static_shell_needs_no_session():
+    """It has to load before any JavaScript can ask for one."""
+    assert check(path="/app/index.html", cookie_session=None) is None
+
+
+@pytest.mark.parametrize("method", ["POST", "PUT", "PATCH", "DELETE"])
+def test_a_mutation_without_a_valid_csrf_token_is_refused(method):
+    for token in (None, "guessed"):
+        refusal = check(method=method, path="/api/v1/jobs", **{CSRF_HEADER: token})
+        assert (refusal.status, refusal.code) == (403, "csrf_failed")
+
+
+@pytest.mark.parametrize("method", ["GET", "HEAD", "OPTIONS"])
+def test_reads_need_no_csrf_token(method):
+    assert check(method=method, **{CSRF_HEADER: None}) is None
+
+
+def test_an_oversized_body_is_refused_before_it_is_read():
+    refusal = check(method="POST", path="/api/v1/jobs", content_length=str(LIMIT + 1))
+    assert (refusal.status, refusal.code) == (413, "payload_too_large")
+    assert check(method="POST", path="/api/v1/jobs", content_length=str(LIMIT)) is None
+
+
+def test_the_host_check_precedes_authentication():
+    """A refusal must not tell a foreign caller whether its session guess was right."""
+    refusal = check(host="evil.example", cookie_session="guessed")
+    assert refusal.code == "host_not_allowed"
+
+
+def test_refusals_name_no_path_secret_or_internal():
+    refusals = [
+        check(host="evil.example"),
+        check(method="POST", origin="http://evil.example"),
+        check(cookie_session=None),
+        check(method="POST", **{CSRF_HEADER: None}),
+    ]
+    for refusal in refusals:
+        assert SESSION.id not in refusal.message
+        assert SESSION.csrf_token not in refusal.message
+        assert "/" not in refusal.message
+
+
+@pytest.mark.parametrize(
+    "value,expected",
+    [
+        ("127.0.0.1:8000", ("127.0.0.1", "8000")),
+        ("localhost", ("localhost", "")),
+        ("[::1]:8000", ("[::1]", "8000")),
+        ("[::1]", ("[::1]", "")),
+        ("", ("", "")),
+    ],
+)
+def test_split_host(value, expected):
+    assert split_host(value) == expected