US07-03: Harden Media and Metadata Edge Cases

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,99 @@ 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.
+
+## Media and metadata hardening (US07-03)
+
+Every pixel the application reads goes through `photo_pipeline/imaging.py`: the
+declared dimensions are checked before anything is decoded, Pillow's
+decompression-bomb warning is treated as a refusal, JPEG decodes near the requested
+size, and each decoder failure becomes one of two typed errors. A damaged file is a
+per-item error with a persisted code, never a failed scan or a dead worker.
+
+Every metadata stage ends with an EXIF checkpoint (`services/exif_checkpoint.py`):
+snapshot, write the owned keywords, read back, prove the owned fields landed and that
+nothing else moved, refresh the file hash. A field the stage does not own that
+changed anyway makes the checkpoint `divergent` — recorded in `exif_projections`,
+shown in the review queue, never repaired behind the user's back, and not counted as
+verified, so upload stays blocked.
+
+The golden corpus that proves all of it is generated, not committed:
+`tests/fixtures/media_corpus.py` declares every format, orientation, profile,
+damage, and metadata case with its expected outcome, and the suite regenerates it
+twice to prove it does not drift.
+
+```bash
+work_item/scripts/python -m pytest tests/integration/test_media_hardening.py tests/integration/test_exif_checkpoints.py -q
+```
+
+## 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
@@ -438,6 +480,13 @@ rows:
       inference path needs the local model + deterministic fake.
     target: photo_pipeline/integrations/nsfw_model.py
     tests: [test_nsfwtag::test_score_images_cache_hit_skips_model]
+    delta: >
+      The donor set Pillow's process-global ImageFile.LOAD_TRUNCATED_IMAGES so a
+      partially downloaded file still scored. Here the same process also hashes
+      files and renders previews, and those must keep refusing a truncated file
+      rather than silently working on half of one; scoring opens images through
+      the bounded photo_pipeline.imaging door instead and skips the ones it cannot
+      read, leaving them unscored and visibly undecided (US07-03).
     status: characterized
 
   - id: nt-exif-keyword
@@ -485,8 +534,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 +550,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 +564,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 +637,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 +655,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/0015_exif_projections.py

@@ -0,0 +1,42 @@
+"""Durable EXIF projections per asset and stage (US07-03).
+
+Revision ID: 0015_exif_projections
+Revises: 0014_restore_plans
+Create Date: 2026-08-16
+
+The concept's ``exif_projections`` table, added at the point it earns its keep: a
+checkpoint that finds a field it does not own changed must be able to say so after
+a restart. ``state`` is verified | divergent | failed, and only ``verified`` counts
+as a completed metadata stage.
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = "0015_exif_projections"
+down_revision = "0014_restore_plans"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "exif_projections",
+        sa.Column("asset_id", sa.String(), sa.ForeignKey("assets.id"), primary_key=True),
+        sa.Column("stage", sa.String(), primary_key=True),  # safety | analysis
+        sa.Column("id", sa.String(), nullable=False),
+        sa.Column("projection_version", sa.Integer(), nullable=False, server_default="0"),
+        # What the stage asked for: {"add": [...], "remove": [...]}.
+        sa.Column("desired_json", sa.String(), nullable=True),
+        # Fields outside the stage's ownership that did not survive the write.
+        sa.Column("divergent_fields", sa.String(), nullable=True),
+        sa.Column("result_file_sha256", sa.String(), nullable=True),
+        sa.Column("state", sa.String(), nullable=False),
+        sa.Column("error_code", sa.String(), nullable=True),
+        sa.Column("verified_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=True),
+    )
+
+
+def downgrade() -> None:
+    op.drop_table("exif_projections")

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,20 +29,58 @@ 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
 from photo_pipeline.config import Config
 from photo_pipeline.db import create_db_engine, create_session_factory, run_migrations
 from photo_pipeline.logging import configure_logging
+from photo_pipeline.services.thumbnails import ThumbnailService
 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()
@@ -55,6 +97,9 @@ def create_app(config: Config | None = None) -> FastAPI:
         # An upload whose process died left no outcome behind; resolve it now so the
         # uploader lane is free and the uncertain batch is visible (US05-02).
         UploadBatchService(app.state.session_factory, config=config).recover()
+        # A render killed mid-write leaves its temporary beside the cache entry;
+        # remove those recognized leftovers, and only those (US07-03).
+        ThumbnailService(app.state.session_factory, config).cleanup_temp_files()
         try:
             yield
         finally:
@@ -62,6 +107,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/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/imaging.py

@@ -0,0 +1,119 @@
+"""Bounded, defensive image decoding — one door for every pixel this app reads.
+
+A photo library contains files nobody planned for: truncated downloads, zero-byte
+placeholders, a PNG whose header claims 200000×200000, a TIFF with a broken ICC
+profile, an extension that lies about its content. None of them may take down a
+request or a worker, and none may decode more pixels than the caller allowed
+(concept §17: decoded pixels, not file size, are what exhausts memory).
+
+``open_image`` is that single door:
+
+* the declared dimensions are checked **before** a pixel is decoded;
+* Pillow's decompression-bomb *warning* is promoted to an error, because the
+  warning band (between Pillow's limit and twice it) still decodes the image;
+* every decoder failure — at open time or during the caller's decode — becomes one
+  of two typed errors, so callers map them to their own item state instead of
+  catching ``Exception``;
+* error text names no path: it reaches API responses, and the full reason goes to
+  the server log instead (US07-02).
+
+``to_srgb`` and ``draft`` are the other two bounded-decode helpers: colour-manage a
+profile-bearing image into sRGB, and let JPEG decode straight to a size near the
+requested one rather than at full resolution.
+"""
+
+from __future__ import annotations
+
+import io
+import logging
+import warnings
+from contextlib import contextmanager
+from pathlib import Path
+
+from PIL import Image, ImageCms, ImageFile, UnidentifiedImageError
+
+log = logging.getLogger(__name__)
+
+# Matches ``Config.thumbnail_max_pixels``; used where no configuration is at hand
+# (hashing runs inside discovery, which takes no config).
+DEFAULT_MAX_PIXELS = 100_000_000
+
+
+class MediaError(Exception):
+    """A file could not be turned into pixels safely."""
+
+
+class UndecodableImage(MediaError):
+    """Corrupt, truncated, empty, or not an image at all."""
+
+
+class ImageTooLarge(MediaError):
+    """More pixels than this operation is allowed to decode."""
+
+
+@contextmanager
+def open_image(path: Path | str, *, max_pixels: int = DEFAULT_MAX_PIXELS):
+    """Yield an open :class:`PIL.Image.Image`, bounded and with typed failures.
+
+    Decoder errors raised inside the ``with`` body are translated too — a truncated
+    JPEG only fails when its pixels are actually pulled, which is the caller's line,
+    not this one.
+    """
+    # Pillow's truncation tolerance is a process-global switch that any library in
+    # the process can flip (the donor CLI did). This door decides the policy for its
+    # own callers: half a file is not a picture.
+    tolerated = ImageFile.LOAD_TRUNCATED_IMAGES
+    ImageFile.LOAD_TRUNCATED_IMAGES = False
+    with warnings.catch_warnings():
+        # The warning band is not a warning for us: it means Pillow was willing to
+        # decode an image large enough to be a denial-of-service.
+        warnings.simplefilter("error", Image.DecompressionBombWarning)
+        try:
+            with Image.open(path) as image:
+                width, height = image.size
+                if width * height > max_pixels:
+                    raise ImageTooLarge(f"{width}x{height} exceeds the {max_pixels} pixel limit")
+                yield image
+        except MediaError:
+            raise
+        except (Image.DecompressionBombError, Image.DecompressionBombWarning) as error:
+            log.info("refused oversized image %s: %s", path, error)
+            raise ImageTooLarge("image exceeds the decompression-bomb limit") from error
+        except (UnidentifiedImageError, OSError, ValueError, SyntaxError, MemoryError) as error:
+            log.info("cannot decode %s: %s", path, error)
+            raise UndecodableImage(f"cannot decode image ({type(error).__name__})") from error
+        finally:
+            ImageFile.LOAD_TRUNCATED_IMAGES = tolerated
+
+
+def draft(image: Image.Image, size: int) -> None:
+    """Ask the decoder for a smaller image where the format allows it (JPEG).
+
+    This is the difference between decoding a 40-megapixel JPEG and decoding the
+    roughly 1-megapixel version a 1280px preview needs.
+    """
+    try:
+        image.draft(None, (size, size))
+    except (AttributeError, ValueError, OSError):  # not a draft-capable format
+        pass
+
+
+def to_srgb(image: Image.Image, *, mode: str) -> Image.Image:
+    """Convert into ``mode``, colour-managing through an embedded ICC profile.
+
+    Without this a wide-gamut original renders with visibly wrong colours, because
+    its numbers are interpreted as sRGB. A broken or unreadable profile is not a
+    reason to fail a preview — the plain conversion is still a correct picture.
+    """
+    profile = image.info.get("icc_profile")
+    if profile:
+        try:
+            return ImageCms.profileToProfile(
+                image,
+                ImageCms.ImageCmsProfile(io.BytesIO(profile)),
+                ImageCms.createProfile("sRGB"),
+                outputMode=mode,
+            )
+        except Exception as error:  # noqa: BLE001 - any ICC failure falls back
+            log.info("ignoring unusable ICC profile on %s: %s", getattr(image, "filename", "?"), error)
+    return image.convert(mode)

photo_pipeline/integrations/exiftool.py

@@ -56,6 +56,30 @@ def read_keyword_sets(paths: Iterable[str]) -> dict[str, set[str]]:
     return out
 
 
+def read_all(path: str) -> dict | None:
+    """Every tag exiftool can read from ``path``, or ``None`` when it cannot answer.
+
+    This is the snapshot an EXIF checkpoint compares against: proving that a write
+    preserved the fields it does not own requires knowing all of them, not just the
+    ones being written (US07-03). ``None`` (exiftool missing, unreadable file,
+    unparsable output) is not an empty snapshot — a caller must not read it as
+    "nothing was there".
+    """
+    try:
+        result = subprocess.run(
+            ["exiftool", "-m", "-j", "-G0:1", path], capture_output=True, text=True
+        )
+    except FileNotFoundError:
+        return None
+    try:
+        records = json.loads(result.stdout or "[]")
+    except ValueError:
+        return None
+    if not records:
+        return None
+    return {k: v for k, v in records[0].items() if k != "SourceFile"}
+
+
 def apply_keywords(path: str, *, add: Iterable[str] = (), remove: Iterable[str] = ()) -> bool:
     """Idempotently add/remove keywords in Keywords + Subject; preserve all else."""
     args = ["exiftool", "-m", "-overwrite_original"]

photo_pipeline/integrations/nsfw_model.py

@@ -15,6 +15,8 @@ from __future__ import annotations
 
 from pathlib import Path
 
+from photo_pipeline import imaging
+
 MODEL_ID = "AdamCodd/vit-base-nsfw-detector"
 BATCH = 16
 
@@ -54,13 +56,17 @@ class NsfwModel:
         self._ensure_loaded()
         import numpy as np
         import torch
-        from PIL import Image, ImageFile
+        from PIL import Image
 
-        ImageFile.LOAD_TRUNCATED_IMAGES = True
-
-        def preprocess(image):
-            image = image.convert("RGB").resize((self._size, self._size), Image.BILINEAR)
-            array = (np.asarray(image, dtype="float32") / 255.0 - 0.5) / 0.5
+        # The donor set ``ImageFile.LOAD_TRUNCATED_IMAGES = True`` here. That flag is
+        # process-global: in this application the same process also hashes files and
+        # renders previews, and those must keep failing loudly on a truncated file
+        # rather than quietly working on half of one (US07-03). An unreadable image
+        # is skipped instead — it stays unscored, and therefore visibly undecided.
+        def preprocess(path):
+            with imaging.open_image(path) as image:
+                small = image.convert("RGB").resize((self._size, self._size), Image.BILINEAR)
+            array = (np.asarray(small, dtype="float32") / 255.0 - 0.5) / 0.5
             return torch.from_numpy(array).permute(2, 0, 1)
 
         results: list[tuple[str, float]] = []
@@ -69,9 +75,10 @@ class NsfwModel:
             tensors, batch_paths = [], []
             for path in items[start : start + self.batch]:
                 try:
-                    tensors.append(preprocess(Image.open(path)))
+                    tensors.append(preprocess(path))
                     batch_paths.append(path)
-                except Exception:
+                except (imaging.MediaError, OSError, ValueError):
+                    # One bad file must not cost the batch its other fifteen.
                     continue
             if not tensors:
                 continue

photo_pipeline/jobs/domain_handlers.py

@@ -38,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:

photo_pipeline/models/__init__.py

@@ -7,6 +7,7 @@ Alembic environment relies on.
 from photo_pipeline.models.albums import AlbumProposal
 from photo_pipeline.models.archives import ArchiveLocation, ArchiveOperation, ArchivePlan
 from photo_pipeline.models.assets import Asset, AssetPath
+from photo_pipeline.models.exif import ExifProjection
 from photo_pipeline.models.duplicates import (
     DuplicateCluster,
     DuplicateMember,
@@ -28,6 +29,7 @@ __all__ = [
     "DuplicateCluster",
     "DuplicateMember",
     "DuplicateNegativeLink",
+    "ExifProjection",
     "Job",
     "JobItem",
     "JobEvent",

photo_pipeline/models/exif.py

@@ -0,0 +1,33 @@
+"""The durable EXIF projection per asset and stage (concept §3, US07-03).
+
+One row per ``(asset_id, stage)``: what the stage wanted written, what the file
+looked like afterwards, and whether anything outside the stage's ownership moved.
+``state = divergent`` is the whole point of the table — it survives restarts, keeps
+the asset out of stages that require verified metadata, and gives a human something
+to look at instead of a silent repair.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime
+
+from sqlalchemy import DateTime, ForeignKey, Integer, String
+from sqlalchemy.orm import Mapped, mapped_column
+
+from photo_pipeline.db import Base
+
+
+class ExifProjection(Base):
+    __tablename__ = "exif_projections"
+
+    asset_id: Mapped[str] = mapped_column(ForeignKey("assets.id"), primary_key=True)
+    stage: Mapped[str] = mapped_column(String, primary_key=True)  # safety | analysis
+    id: Mapped[str] = mapped_column(String, nullable=False)
+    projection_version: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
+    desired_json: Mapped[str | None] = mapped_column(String)
+    divergent_fields: Mapped[str | None] = mapped_column(String)  # JSON array
+    result_file_sha256: Mapped[str | None] = mapped_column(String)
+    state: Mapped[str] = mapped_column(String, nullable=False)  # verified|divergent|failed
+    error_code: Mapped[str | None] = mapped_column(String)
+    verified_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+    updated_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))

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,9 +25,9 @@ from typing import Protocol
 from sqlalchemy import func, select
 from sqlalchemy.orm import sessionmaker
 
-from photo_pipeline.integrations import exiftool
+from photo_pipeline import path_policy
 from photo_pipeline.models import AnalysisResult, Asset, SafetyReview
-from photo_pipeline.services import hashing
+from photo_pipeline.services import exif_checkpoint
 from photo_pipeline.services.safety import SFW
 
 MODEL = "gemini-2.5-flash"
@@ -59,9 +59,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 +143,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
@@ -150,7 +169,7 @@ class AnalysisService:
                 tokens=int(result.get("_tokens", 0)) if isinstance(result, dict) else 0,
                 raw=json.dumps(result, ensure_ascii=False),
             )
-            _write_analysis_exif(path, result)
+            self._write_analysis_exif(asset_id, path, result)
             analyzed += 1
         return {"analyzed": analyzed, "skipped": skipped, "errors": errors}
 
@@ -175,10 +194,42 @@ class AnalysisService:
                 row.location_hint = result.get("location_hint")
                 row.approx_year = result.get("approx_year")
                 row.analyzed_at = now
-                row.exif_written_at = now
             session.add(row)
             session.commit()
 
+    def _write_analysis_exif(self, asset_id: str, path: str, result: dict) -> None:
+        """The analysis EXIF checkpoint: additive keywords, then prove the rest held.
+
+        Additive by design — safety keywords and the user's own keywords are merged
+        with, never replaced (concept §3). ``exif_written_at`` is set only when the
+        read-back verified both the new keywords and every field this stage does not
+        own; a divergent result is recorded and left for a human (US07-03).
+        ponytail: the managed ``AI:`` caption segment, once captions are owned here.
+        """
+        tags = tuple(str(tag) for tag in (result.get("tags") or []))
+        if not tags:
+            return
+        checkpoint = exif_checkpoint.run(path, add=tags)
+        exif_checkpoint.record(
+            self._session_factory,
+            asset_id=asset_id,
+            stage="analysis",
+            result=checkpoint,
+            add=tags,
+        )
+        if not checkpoint.verified:
+            return
+        with self._session_factory() as session:
+            row = session.get(AnalysisResult, asset_id)
+            if row is not None:
+                row.exif_written_at = checkpoint.verified_at
+            asset = session.get(Asset, asset_id)
+            if asset is not None and checkpoint.sha256:
+                # The bytes changed when the container was rewritten; upload must use
+                # the hash of what is actually on disk now (concept §3).
+                asset.current_sha256 = checkpoint.sha256
+            session.commit()
+
     def get(self, asset_id: str) -> dict | None:
         with self._session_factory() as session:
             row = session.get(AnalysisResult, asset_id)
@@ -191,17 +242,6 @@ def _album_hint(path: str) -> str:
     return Path(path).parent.name
 
 
-def _write_analysis_exif(path: str, result: dict) -> None:
-    """Additive analysis keywords into EXIF (Keywords/Subject), preserving safety
-    and user keywords. The donor also wrote a managed caption; only keywords are
-    written here via the shared adapter.
-    ponytail: add the managed ``AI:`` caption segment + read-back preservation check
-    when the analysis EXIF checkpoint is hardened."""
-    tags = result.get("tags") if isinstance(result, dict) else None
-    if tags:
-        exiftool.apply_keywords(path, add=[str(t) for t in tags])
-
-
 def _result_dict(row: AnalysisResult) -> dict:
     data = {field: getattr(row, field) for field in RESULT_FIELDS}
     data["tags"] = json.loads(row.tags) if row.tags else []

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/exif_checkpoint.py

@@ -0,0 +1,173 @@
+"""The EXIF checkpoint every metadata stage ends with (concept §3, US07-03).
+
+A stage does not own a file's metadata; it owns a few fields in it. So writing is
+never "set these tags" — it is:
+
+    snapshot everything → write only the owned fields → read everything back
+    → prove the owned fields landed → prove nothing else moved
+    → refresh the file hash → record the projection
+
+Non-destructive here means *semantic* preservation: exiftool may rewrite the whole
+container, so the file's bytes, size, and timestamps legitimately change. What may
+not change is any field this stage does not own. When one does, the checkpoint is
+``divergent``: the result is recorded, the stage is **not** marked verified, and
+nothing is silently repaired — a later stage that needs verified metadata (upload)
+therefore stays blocked until a human looks.
+
+``failed`` is the third outcome and is deliberately distinct: exiftool missing, an
+unreadable file, or a write that did not take is not evidence that metadata is fine.
+"""
+
+from __future__ import annotations
+
+import json
+import uuid
+from dataclasses import dataclass
+from datetime import datetime, timezone
+
+from photo_pipeline.integrations import exiftool
+from photo_pipeline.models import ExifProjection
+from photo_pipeline.services import hashing
+
+# The tags the safety and analysis stages may change. Matched on the tag name, so
+# every group carries the same rule (IPTC:Keywords, XMP:XMP-dc:Subject, ...).
+OWNED_TAGS = frozenset({"Keywords", "Subject"})
+
+# Not metadata about the picture: filesystem facts, the digest that necessarily
+# moves whenever IPTC does, and the structural tags exiftool has to create the first
+# time it writes an IPTC or XMP block. Comparing these would report every write as
+# divergent and make the signal worthless.
+VOLATILE_PREFIXES = ("File:System:", "ExifTool:")
+VOLATILE_KEYS = frozenset(
+    {
+        "File:CurrentIPTCDigest",
+        "IPTC:ApplicationRecordVersion",
+        "XMP:XMP-x:XMPToolkit",
+        "XMP:XMP-xmp:MetadataDate",
+    }
+)
+
+VERIFIED = "verified"
+DIVERGENT = "divergent"
+FAILED = "failed"
+
+
+@dataclass(frozen=True)
+class CheckpointResult:
+    state: str  # verified | divergent | failed
+    changed_fields: tuple[str, ...] = ()
+    sha256: str | None = None
+    verified_at: datetime | None = None
+    reason: str | None = None
+
+    @property
+    def verified(self) -> bool:
+        return self.state == VERIFIED
+
+
+def _now() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+def is_owned(key: str) -> bool:
+    return key.rsplit(":", 1)[-1] in OWNED_TAGS
+
+
+def is_volatile(key: str) -> bool:
+    return key in VOLATILE_KEYS or key.startswith(VOLATILE_PREFIXES)
+
+
+def compare(before: dict, after: dict) -> tuple[str, ...]:
+    """Fields outside this stage's ownership whose value did not survive the write.
+
+    Additions count: a tag that appears out of nowhere is as much a divergence as a
+    tag that disappeared — both mean the write did more than it was asked to.
+    """
+    keys = set(before) | set(after)
+    return tuple(
+        sorted(
+            key
+            for key in keys
+            if not is_owned(key)
+            and not is_volatile(key)
+            and before.get(key) != after.get(key)
+        )
+    )
+
+
+def owned_values(snapshot: dict) -> set[str]:
+    """Lowercased Keywords/Subject values across every group in the snapshot."""
+    values: set[str] = set()
+    for key, value in snapshot.items():
+        if not is_owned(key):
+            continue
+        items = value if isinstance(value, list) else [value]
+        values.update(str(item).strip().lower() for item in items if item is not None)
+    return values
+
+
+def run(
+    path: str,
+    *,
+    add: tuple[str, ...] = (),
+    remove: tuple[str, ...] = (),
+) -> CheckpointResult:
+    """Write the owned keywords for one asset and verify the whole file around them."""
+    before = exiftool.read_all(path)
+    if before is None:
+        return CheckpointResult(FAILED, reason="metadata_unreadable")
+
+    if not exiftool.apply_keywords(path, add=add, remove=remove):
+        return CheckpointResult(FAILED, reason="write_failed")
+
+    after = exiftool.read_all(path)
+    if after is None:
+        return CheckpointResult(FAILED, reason="readback_unreadable")
+
+    present = owned_values(after)
+    wanted = {value.strip().lower() for value in add}
+    unwanted = {value.strip().lower() for value in remove}
+    if not wanted <= present or (unwanted & present):
+        return CheckpointResult(FAILED, reason="owned_fields_not_written")
+
+    changed = compare(before, after)
+    sha256 = hashing.sha256_file(path)
+    if changed:
+        return CheckpointResult(DIVERGENT, changed_fields=changed, sha256=sha256)
+    return CheckpointResult(VERIFIED, sha256=sha256, verified_at=_now())
+
+
+def record(
+    session_factory,
+    *,
+    asset_id: str,
+    stage: str,
+    result: CheckpointResult,
+    add: tuple[str, ...] = (),
+    remove: tuple[str, ...] = (),
+) -> None:
+    """Persist the projection for ``(asset_id, stage)`` — one current row per pair.
+
+    The row is what makes divergence durable and reviewable rather than a log line
+    that scrolled away.
+    """
+    with session_factory() as session:
+        row = session.get(ExifProjection, (asset_id, stage))
+        if row is None:
+            row = ExifProjection(asset_id=asset_id, stage=stage, id=str(uuid.uuid4()))
+            session.add(row)
+        row.projection_version = (row.projection_version or 0) + 1
+        row.desired_json = json.dumps({"add": list(add), "remove": list(remove)})
+        row.divergent_fields = json.dumps(list(result.changed_fields))
+        row.result_file_sha256 = result.sha256
+        row.state = result.state
+        row.error_code = result.reason
+        row.verified_at = result.verified_at
+        row.updated_at = _now()
+        session.commit()
+
+
+def state_for(session_factory, asset_id: str, stage: str) -> str | None:
+    with session_factory() as session:
+        row = session.get(ExifProjection, (asset_id, stage))
+        return row.state if row else None

photo_pipeline/services/hashing.py

@@ -16,7 +16,9 @@ from __future__ import annotations
 import hashlib
 from pathlib import Path
 
-from PIL import Image, ImageOps
+from PIL import ImageOps
+
+from photo_pipeline import imaging
 
 PIXEL_HASH_VERSION = 1
 PHASH_VERSION = 1
@@ -41,7 +43,9 @@ def _digest_file(path: Path | str, digest) -> str:
 
 
 def pixel_sha256(path: Path | str) -> str:
-    with Image.open(path) as image:
+    # Bounded decode: a header claiming a billion pixels is refused before it is
+    # ever turned into memory (US07-03).
+    with imaging.open_image(path) as image:
         oriented = ImageOps.exif_transpose(image)
         rgb = oriented.convert("RGB")
         header = f"{PIXEL_HASH_VERSION}:{rgb.width}x{rgb.height}:".encode()
@@ -68,7 +72,9 @@ def phash(path: Path | str) -> str:
     import numpy as np
     from scipy.fftpack import dct
 
-    with Image.open(path) as image:
+    from PIL import Image
+
+    with imaging.open_image(path) as image:
         small = image.convert("L").resize((32, 32), Image.LANCZOS)
         matrix = np.asarray(small, dtype=np.float64)
     transformed = dct(dct(matrix, axis=0), axis=1)

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

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

photo_pipeline/services/safety.py

@@ -108,9 +108,8 @@ from datetime import datetime, timezone
 from sqlalchemy import select
 from sqlalchemy.orm import sessionmaker
 
-from photo_pipeline.integrations import exiftool
-from photo_pipeline.models import Asset, SafetyReview
-from photo_pipeline.services import hashing
+from photo_pipeline.models import Asset, ExifProjection, SafetyReview
+from photo_pipeline.services import exif_checkpoint
 
 DECISIONS = {SFW, NSFW, "deferred"}
 
@@ -174,6 +173,14 @@ class SafetyService:
         with self._session_factory() as session:
             assets = list(session.scalars(_eligible_assets_query().order_by(Asset.current_path)))
             latest = self._latest_by_asset(session)
+            # One query, not one per asset: the reviewer needs to see a divergent
+            # checkpoint, which is neither "verified" nor a plain failure (US07-03).
+            projections = {
+                row.asset_id: row.state
+                for row in session.scalars(
+                    select(ExifProjection).where(ExifProjection.stage == "safety")
+                )
+            }
             rows = []
             for asset in assets:
                 review = latest.get(asset.id)
@@ -189,6 +196,7 @@ class SafetyService:
                         "decision": decision,
                         "suggested": classify(review.score) if review and review.score is not None else None,
                         "exif_verified": bool(review and review.exif_verified_at),
+                        "exif_state": projections.get(asset.id),
                     }
                 )
             return {"total": len(rows), "items": rows[offset : offset + limit]}
@@ -247,13 +255,24 @@ class SafetyService:
         result_sha256 = None
         if write_exif and decision in (SFW, NSFW) and path:
             ops = exif_projection(decision)
-            if exiftool.apply_keywords(path, add=ops["add"], remove=ops["remove"]):
-                # Read back: the chosen keyword present, the opposite absent.
-                keywords = exiftool.read_keyword_sets([path]).get(path, set())
-                opposite = NSFW if decision == SFW else SFW
-                if decision in keywords and opposite not in keywords:
-                    exif_verified_at = _now()
-                    result_sha256 = hashing.sha256_file(path)
+            # The full checkpoint: write the owned keyword, read the whole file back,
+            # and prove every field this stage does not own survived. A divergent
+            # result is recorded and left alone — it must not count as verified, so
+            # upload stays blocked until a human decides (US07-03).
+            result = exif_checkpoint.run(
+                path, add=tuple(ops["add"]), remove=tuple(ops["remove"])
+            )
+            exif_checkpoint.record(
+                self._session_factory,
+                asset_id=asset_id,
+                stage="safety",
+                result=result,
+                add=tuple(ops["add"]),
+                remove=tuple(ops["remove"]),
+            )
+            if result.verified:
+                exif_verified_at = result.verified_at
+                result_sha256 = result.sha256
 
         now = _now()
         with self._session_factory() as session:

photo_pipeline/services/thumbnails.py

@@ -20,15 +20,16 @@ pa-imaging).
 
 from __future__ import annotations
 
+import os
 import uuid
 from pathlib import Path
 
-from PIL import Image, ImageOps, UnidentifiedImageError
+from PIL import Image, ImageOps
 from sqlalchemy import select
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import sessionmaker
 
-from photo_pipeline import path_policy
+from photo_pipeline import imaging, path_policy
 from photo_pipeline.config import Config
 from photo_pipeline.models import Asset, Thumbnail
 from photo_pipeline.services import availability
@@ -42,6 +43,9 @@ except Exception:  # pragma: no cover
     pass
 
 SIZES = (256, 512, 1280)
+# Every in-flight render writes one of these beside its destination; startup
+# cleanup recognises exactly this pattern and nothing else.
+TEMP_SUFFIX = ".tmp"
 THUMB_VERSION = 1
 THUMB_FORMAT = "webp"
 # The size kept as durable comparison evidence for archived assets (concept §9).
@@ -129,7 +133,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 +202,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
@@ -223,28 +226,33 @@ class ThumbnailService:
         return self._cache_dir / safe[:2] / f"{safe}.{THUMB_FORMAT}"
 
     def _render(self, source: str, size: int, cache_key: str) -> dict:
+        """Decode bounded, orient, colour-manage, resize, and write atomically.
+
+        The temporary file is removed on every failure path: a decoder that dies
+        halfway through ``save`` would otherwise leave a stray ``.tmp`` in the cache
+        forever (US07-03).
+        """
+        destination = self._cache_path(cache_key)
+        destination.parent.mkdir(parents=True, exist_ok=True)
+        tmp = destination.with_name(f".{destination.name}.{uuid.uuid4().hex}{TEMP_SUFFIX}")
         try:
-            with Image.open(source) as image:
-                width, height = image.size
-                if width * height > self._config.thumbnail_max_pixels:
-                    raise ImageTooLarge(
-                        f"{width}x{height} exceeds {self._config.thumbnail_max_pixels} px"
-                    )
+            with imaging.open_image(source, max_pixels=self._config.thumbnail_max_pixels) as image:
+                imaging.draft(image, size)  # JPEG decodes near the target size
                 oriented = ImageOps.exif_transpose(image)
                 mode = "RGBA" if _has_alpha(oriented) else "RGB"
-                converted = oriented.convert(mode)
+                converted = imaging.to_srgb(oriented, mode=mode)
                 converted.thumbnail((size, size), Image.LANCZOS)
                 out_width, out_height = converted.size
-                destination = self._cache_path(cache_key)
-                destination.parent.mkdir(parents=True, exist_ok=True)
-                tmp = destination.with_name(f".{destination.name}.{uuid.uuid4().hex}.tmp")
                 converted.save(tmp, format="WEBP", quality=82, method=4)
-        except ImageTooLarge:
+        except imaging.ImageTooLarge as error:
+            tmp.unlink(missing_ok=True)
+            raise ImageTooLarge(str(error)) from None
+        except imaging.UndecodableImage as error:
+            tmp.unlink(missing_ok=True)
+            raise UnsupportedImage(str(error)) from None
+        except Exception:
+            tmp.unlink(missing_ok=True)
             raise
-        except (UnidentifiedImageError, OSError, ValueError) as error:
-            raise UnsupportedImage(f"cannot decode {source}: {error}") from error
-
-        import os
 
         os.replace(tmp, destination)
         return {
@@ -254,6 +262,28 @@ class ThumbnailService:
             "format": THUMB_FORMAT,
         }
 
+    def cleanup_temp_files(self) -> int:
+        """Remove leftover render temporaries, and only those.
+
+        Runs at startup, where the concept is explicit: clean *recognized* stale
+        temporary files, never arbitrary ones. The pattern is this service's own —
+        a dot-prefixed name inside the managed cache directory ending in
+        ``.tmp`` — so nothing outside the cache and nothing a user put there can
+        match. Returns how many were removed.
+        """
+        if not self._cache_dir.is_dir():
+            return 0
+        removed = 0
+        for leftover in self._cache_dir.rglob(f".*{TEMP_SUFFIX}"):
+            if not leftover.is_file() or leftover.is_symlink():
+                continue
+            try:
+                leftover.unlink()
+                removed += 1
+            except OSError:  # pragma: no cover - a racing render already won
+                pass
+        return removed
+
     # ── persistence ────────────────────────────────────────────────────────────
     def _record_ready(
         self, cache_key: str, asset_id: str, size: int, rendered: dict, *, protected: bool = False

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/fixtures/media_corpus.py

@@ -0,0 +1,325 @@
+"""The golden media corpus: every format, orientation, profile, damage, and
+metadata case this application claims to survive (US07-03, concept §18).
+
+``CASES`` is the manifest and the authority. Each entry declares a stable logical
+id (never a path), how the file is generated, and what the pipeline must do with
+it — decode it, or refuse it with one precise error code. Tests parametrize over the
+manifest, so a case that is added here without an expectation, or an expectation
+that stops holding, fails the suite rather than quietly going untested.
+
+Everything is generated, never committed: fixed pixel seeds, fixed EXIF strings, no
+clock, no network, no personal data. Regeneration is byte-stable, which
+``test_media_hardening.py`` proves by building the corpus twice and comparing
+checksums — a golden corpus that drifts is not golden.
+"""
+
+from __future__ import annotations
+
+import struct
+import zlib
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Callable
+
+import numpy as np
+from PIL import Image
+
+# Error codes the pipeline may answer with; ``None`` means "must render".
+UNSUPPORTED = "unsupported_image"
+TOO_LARGE = "image_too_large"
+
+# One fixed capture time for every metadata case: the corpus must not depend on when
+# it was generated.
+CAPTURE_TIME = "2019:07:14 10:30:00"
+
+
+@dataclass(frozen=True)
+class MediaCase:
+    id: str
+    filename: str
+    kind: str  # format | orientation | profile | damaged | metadata
+    build: Callable[[Path], None]
+    expect_error: str | None = None
+    # Declared for the metadata cases: exiftool arguments applied after the pixels
+    # are written, and the user fields that must survive every later stage.
+    exif_args: tuple[str, ...] = ()
+    preserved_fields: tuple[str, ...] = ()
+    notes: str = ""
+    tags: tuple[str, ...] = field(default_factory=tuple)
+
+
+# ── generators ────────────────────────────────────────────────────────────────
+
+
+def _pixels(width: int, height: int, seed: int, bands: int = 3) -> np.ndarray:
+    return np.random.default_rng(seed).integers(0, 256, (height, width, bands), dtype=np.uint8)
+
+
+def _save(path: Path, image: Image.Image, **kwargs) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    image.save(path, **kwargs)
+
+
+def _jpeg(width=320, height=240, seed=1, **kwargs):
+    def build(path: Path) -> None:
+        _save(path, Image.fromarray(_pixels(width, height, seed)), quality=90, **kwargs)
+
+    return build
+
+
+def _oriented_jpeg(orientation: int):
+    def build(path: Path) -> None:
+        image = Image.fromarray(_pixels(400, 200, 5))  # landscape source
+        exif = image.getexif()
+        exif[274] = orientation  # 0x0112 Orientation
+        _save(path, image, exif=exif, quality=90)
+
+    return build
+
+
+def _rotated_pixels(path: Path) -> None:
+    """The same scene rotated in the pixels instead of in a tag."""
+    _save(path, Image.fromarray(_pixels(400, 200, 5)).transpose(Image.ROTATE_90), quality=90)
+
+
+def _png(alpha: bool = False, seed: int = 2):
+    def build(path: Path) -> None:
+        if alpha:
+            _save(path, Image.fromarray(_pixels(120, 90, seed, bands=4), "RGBA"))
+        else:
+            _save(path, Image.fromarray(_pixels(120, 90, seed)))
+
+    return build
+
+
+def _webp(path: Path) -> None:
+    _save(path, Image.fromarray(_pixels(150, 100, 3)), quality=80)
+
+
+def _tiff(path: Path) -> None:
+    _save(path, Image.fromarray(_pixels(140, 110, 4)))
+
+
+def _grayscale(path: Path) -> None:
+    _save(path, Image.fromarray(_pixels(100, 100, 6)).convert("L"), quality=90)
+
+
+def _cmyk(path: Path) -> None:
+    _save(path, Image.fromarray(_pixels(100, 100, 7)).convert("CMYK"), quality=90)
+
+
+def _tiny(path: Path) -> None:
+    _save(path, Image.fromarray(_pixels(1, 1, 8)))
+
+
+def _icc_tagged(path: Path) -> None:
+    """A profile-bearing image: the colour-managed decode path must run."""
+    from PIL import ImageCms
+
+    profile = bytearray(ImageCms.ImageCmsProfile(ImageCms.createProfile("sRGB")).tobytes())
+    # An ICC header carries its creation timestamp at bytes 24..35. Left alone, the
+    # corpus would be a different corpus every time it is generated.
+    profile[24:36] = b"\x00" * 12
+    _save(path, Image.fromarray(_pixels(120, 80, 9)), icc_profile=bytes(profile), quality=90)
+
+
+def _broken_icc(path: Path) -> None:
+    """A profile that is not a profile: a picture is still a picture."""
+    _save(path, Image.fromarray(_pixels(120, 80, 10)), icc_profile=b"not-a-profile", quality=90)
+
+
+def _wide_jpeg(path: Path) -> None:
+    """Large enough that decoding it at full resolution is visible in memory."""
+    _save(path, Image.fromarray(_pixels(4000, 3000, 11)), quality=70)
+
+
+def _zero_byte(path: Path) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_bytes(b"")
+
+
+def _truncated_jpeg(path: Path) -> None:
+    image = Image.fromarray(_pixels(400, 300, 12))
+    _save(path, image, quality=90)
+    data = path.read_bytes()
+    path.write_bytes(data[: len(data) // 2])  # header intact, pixels missing
+
+
+def _corrupt_png(path: Path) -> None:
+    image = Image.fromarray(_pixels(120, 90, 13))
+    _save(path, image)
+    data = bytearray(path.read_bytes())
+    data[40:80] = b"\x00" * 40  # shred the compressed stream, keep the header
+    path.write_bytes(bytes(data))
+
+
+def _not_an_image(path: Path) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_bytes(b"This is a text file that happens to be named .jpg\n")
+
+
+def _png_declaring(width: int, height: int):
+    """A tiny, structurally valid PNG whose header claims an enormous picture.
+
+    A few hundred bytes on disk, gigapixels on paper: the pipeline must refuse it
+    from the declared dimensions, before a single pixel is allocated. Rewriting the
+    IHDR of a real PNG (rather than hand-rolling a stub) keeps the file openable, so
+    the refusal is proven to come from the size check and not from a parse failure.
+    """
+
+    def build(path: Path) -> None:
+        import io
+
+        buffer = io.BytesIO()
+        Image.fromarray(_pixels(4, 4, 15)).save(buffer, format="PNG")
+        data = bytearray(buffer.getvalue())
+        start = 8 + 4  # PNG signature, then the IHDR length field
+        struct.pack_into(">II", data, start + 4, width, height)
+        ihdr = bytes(data[start : start + 4 + 13])
+        struct.pack_into(">I", data, start + 4 + 13, zlib.crc32(ihdr))
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_bytes(bytes(data))
+
+    return build
+
+
+CASES: tuple[MediaCase, ...] = (
+    # ── formats ───────────────────────────────────────────────────────────────
+    MediaCase("jpeg", "formats/plain.jpg", "format", _jpeg()),
+    MediaCase("jpeg_uppercase_ext", "formats/UPPER.JPG", "format", _jpeg(seed=14)),
+    MediaCase("png", "formats/plain.png", "format", _png()),
+    MediaCase("png_alpha", "formats/alpha.png", "format", _png(alpha=True)),
+    MediaCase("webp", "formats/plain.webp", "format", _webp),
+    MediaCase("tiff", "formats/plain.tiff", "format", _tiff),
+    # ── orientation ───────────────────────────────────────────────────────────
+    *(
+        MediaCase(
+            f"orientation_{value}",
+            f"orientation/exif_{value}.jpg",
+            "orientation",
+            _oriented_jpeg(value),
+            notes="EXIF orientation must be applied before resizing",
+        )
+        for value in range(1, 9)
+    ),
+    MediaCase("rotated_pixels", "orientation/rotated.jpg", "orientation", _rotated_pixels),
+    # ── colour and size profiles ──────────────────────────────────────────────
+    MediaCase("grayscale", "profiles/gray.jpg", "profile", _grayscale),
+    MediaCase("cmyk", "profiles/cmyk.jpg", "profile", _cmyk),
+    MediaCase("tiny", "profiles/tiny.png", "profile", _tiny),
+    MediaCase("icc_tagged", "profiles/icc.jpg", "profile", _icc_tagged),
+    MediaCase(
+        "broken_icc",
+        "profiles/broken-icc.jpg",
+        "profile",
+        _broken_icc,
+        notes="an unusable ICC profile falls back to a plain conversion, never an error",
+    ),
+    MediaCase(
+        "large_jpeg",
+        "profiles/large.jpg",
+        "profile",
+        _wide_jpeg,
+        notes="12 megapixels: the decode must stay near the requested size",
+    ),
+    # ── damaged and hostile inputs ────────────────────────────────────────────
+    MediaCase("zero_byte", "damaged/empty.jpg", "damaged", _zero_byte, UNSUPPORTED),
+    MediaCase("truncated_jpeg", "damaged/truncated.jpg", "damaged", _truncated_jpeg, UNSUPPORTED),
+    MediaCase("corrupt_png", "damaged/corrupt.png", "damaged", _corrupt_png, UNSUPPORTED),
+    MediaCase("text_as_jpeg", "damaged/text.jpg", "damaged", _not_an_image, UNSUPPORTED),
+    MediaCase(
+        "bomb_header",
+        "damaged/bomb.png",
+        "damaged",
+        _png_declaring(60_000, 60_000),
+        TOO_LARGE,
+        notes="3.6 gigapixels declared in the header and nothing else",
+    ),
+    MediaCase(
+        "bomb_warning_band",
+        "damaged/bomb-warning.png",
+        "damaged",
+        _png_declaring(10_000, 10_000),
+        TOO_LARGE,
+        notes="inside Pillow's warn-only band; the warning is promoted to a refusal",
+    ),
+    # ── metadata ──────────────────────────────────────────────────────────────
+    MediaCase("no_exif", "metadata/bare.jpg", "metadata", _jpeg(seed=20)),
+    MediaCase(
+        "user_exif",
+        "metadata/user.jpg",
+        "metadata",
+        _jpeg(seed=21),
+        exif_args=(
+            "-Artist=Ada Lovelace",
+            "-Copyright=(c) Ada",
+            f"-DateTimeOriginal={CAPTURE_TIME}",
+            "-GPSLatitude=48.137",
+            "-GPSLatitudeRef=N",
+            "-Rating=4",
+            "-ImageDescription=A day out",
+        ),
+        preserved_fields=(
+            "EXIF:IFD0:Artist",
+            "EXIF:IFD0:Copyright",
+            "EXIF:ExifIFD:DateTimeOriginal",
+            "EXIF:IFD0:ImageDescription",
+            "XMP:XMP-xmp:Rating",
+        ),
+        notes="user metadata that every stage must leave exactly as it found it",
+    ),
+    MediaCase(
+        "prior_safety_keyword",
+        "metadata/prior-safety.jpg",
+        "metadata",
+        _jpeg(seed=22),
+        exif_args=("-Keywords+=nsfw", "-Subject+=nsfw", "-Artist=Ada Lovelace"),
+        preserved_fields=("EXIF:IFD0:Artist",),
+        notes="a safety decision already written by an earlier run",
+    ),
+    MediaCase(
+        "prior_analysis_keywords",
+        "metadata/prior-analysis.jpg",
+        "metadata",
+        _jpeg(seed=23),
+        exif_args=("-Keywords+=beach", "-Keywords+=sunset", "-Subject+=beach", "-Subject+=sunset"),
+        notes="analysis keywords from an earlier run; a safety write must not drop them",
+    ),
+    MediaCase(
+        "conflicting_safety_keywords",
+        "metadata/conflicting.jpg",
+        "metadata",
+        _jpeg(seed=24),
+        exif_args=("-Keywords+=sfw", "-Keywords+=nsfw", "-Subject+=sfw", "-Subject+=nsfw"),
+        notes="both safety keywords at once: mutually exclusive means one must go",
+    ),
+    MediaCase(
+        "malformed_metadata",
+        "metadata/malformed.jpg",
+        "metadata",
+        _jpeg(seed=25, exif=b"\x00\x01\x02not-a-valid-exif-block"),
+        notes="a broken EXIF block must not stop the picture from being usable",
+    ),
+)
+
+CASES_BY_ID = {case.id: case for case in CASES}
+
+
+def build_corpus(root: Path, *, ids: tuple[str, ...] | None = None) -> dict[str, Path]:
+    """Generate the corpus (or a named subset) under ``root``; return id → path."""
+    import subprocess
+
+    built: dict[str, Path] = {}
+    for case in CASES:
+        if ids is not None and case.id not in ids:
+            continue
+        path = root / case.filename
+        case.build(path)
+        if case.exif_args:
+            subprocess.run(
+                ["exiftool", "-m", "-overwrite_original", *case.exif_args, str(path)],
+                capture_output=True,
+                check=False,
+            )
+        built[case.id] = path
+    return built

tests/integration/test_exif_checkpoints.py

@@ -0,0 +1,297 @@
+"""US07-03: EXIF checkpoints, asserted with before/after metadata snapshots.
+
+Every test here reads the complete metadata of a real file through exiftool before
+the stage runs and again afterwards, then compares the two. That is the only way to
+prove the property the concept actually asks for: a stage owns a few fields and must
+leave literally everything else — dates, GPS, camera, artist, rating, other people's
+keywords — exactly as it found them.
+
+The other half is divergence. When something outside the stage's ownership does move,
+the checkpoint must say so, refuse to call itself verified, and change nothing back:
+a silent repair is how a library quietly loses the user's metadata.
+"""
+
+from __future__ import annotations
+
+import json
+import shutil
+import subprocess
+import uuid
+from datetime import datetime, timezone
+from types import SimpleNamespace
+
+import pytest
+
+from photo_pipeline.config import Config
+from photo_pipeline.db import create_db_engine, create_session_factory, run_migrations
+from photo_pipeline.integrations import exiftool
+from photo_pipeline.models import Asset, ExifProjection
+from photo_pipeline.services import exif_checkpoint
+from photo_pipeline.services.analysis import AnalysisService
+from photo_pipeline.services.safety import SafetyService
+from tests.fixtures.media_corpus import CASES_BY_ID, build_corpus
+
+pytestmark = pytest.mark.skipif(
+    shutil.which("exiftool") is None, reason="exiftool not installed"
+)
+
+
+@pytest.fixture
+def env(tmp_path):
+    data = tmp_path / "data"
+    data.mkdir()
+    lib = tmp_path / "lib"
+    files = build_corpus(
+        lib,
+        ids=(
+            "user_exif",
+            "prior_safety_keyword",
+            "prior_analysis_keywords",
+            "conflicting_safety_keywords",
+            "malformed_metadata",
+            "no_exif",
+        ),
+    )
+    config = Config.from_env(
+        {"PHOTO_PIPELINE_DATA_DIR": str(data), "PHOTO_PIPELINE_LIBRARY_ROOTS": str(lib)}
+    )
+    run_migrations(config.database_url)
+    engine = create_db_engine(config.database_url)
+    sf = create_session_factory(engine)
+    assets = {}
+    with sf() as session:
+        for case_id, path in files.items():
+            asset = Asset(
+                id=str(uuid.uuid4()),
+                original_path=str(path),
+                current_path=str(path),
+                discovered_at=datetime.now(timezone.utc),
+                hash_version=1,
+            )
+            session.add(asset)
+            assets[case_id] = asset.id
+        session.commit()
+    yield SimpleNamespace(config=config, lib=lib, sf=sf, files=files, assets=assets)
+    engine.dispose()
+
+
+class Provider:
+    """A vision provider whose tags are fixed, so the EXIF assertion is the test."""
+
+    def __init__(self, tags):
+        self.tags = list(tags)
+
+    def analyze(self, path, *, album_hint):
+        return {"description": "a photo", "tags": self.tags}
+
+
+def snapshot(path):
+    return exiftool.read_all(str(path))
+
+
+def stage_fields(before, after):
+    """Everything that changed except this application's owned and volatile tags."""
+    return exif_checkpoint.compare(before, after)
+
+
+# ── preservation ──────────────────────────────────────────────────────────────
+
+
+def test_a_safety_decision_preserves_every_user_field(env):
+    case = CASES_BY_ID["user_exif"]
+    path = env.files["user_exif"]
+    before = snapshot(path)
+    assert before, "the fixture must actually carry user metadata"
+
+    result = SafetyService(env.sf).decide(env.assets["user_exif"], "nsfw")
+    after = snapshot(path)
+
+    assert result["exif_verified"] is True
+    assert stage_fields(before, after) == ()
+    for field in case.preserved_fields:
+        assert after[field] == before[field], field
+    assert "nsfw" in exif_checkpoint.owned_values(after)
+    assert "sfw" not in exif_checkpoint.owned_values(after)
+
+
+def test_analysis_keywords_are_additive_and_keep_the_safety_decision(env):
+    """The two stages share the Keywords field; the second must merge, not replace."""
+    asset_id = env.assets["prior_analysis_keywords"]
+    path = env.files["prior_analysis_keywords"]
+    SafetyService(env.sf).decide(asset_id, "sfw")
+    before = snapshot(path)
+
+    AnalysisService(
+        env.sf, provider=Provider(["harbour", "boats"]), library_roots=(env.lib,)
+    ).run([asset_id])
+    after = snapshot(path)
+
+    keywords = exif_checkpoint.owned_values(after)
+    assert {"sfw", "beach", "sunset", "harbour", "boats"} <= keywords
+    assert "nsfw" not in keywords
+    assert stage_fields(before, after) == ()
+
+
+def test_flipping_a_safety_decision_removes_only_the_opposite_keyword(env):
+    asset_id = env.assets["prior_analysis_keywords"]
+    path = env.files["prior_analysis_keywords"]
+    safety = SafetyService(env.sf)
+    safety.decide(asset_id, "nsfw")
+    before = snapshot(path)
+
+    safety.decide(asset_id, "sfw")
+    after = snapshot(path)
+
+    keywords = exif_checkpoint.owned_values(after)
+    assert "sfw" in keywords and "nsfw" not in keywords
+    assert {"beach", "sunset"} <= keywords, "analysis keywords are not safety's to remove"
+    assert stage_fields(before, after) == ()
+
+
+def test_conflicting_safety_keywords_are_resolved_to_one(env):
+    asset_id = env.assets["conflicting_safety_keywords"]
+    SafetyService(env.sf).decide(asset_id, "sfw")
+    keywords = exif_checkpoint.owned_values(snapshot(env.files["conflicting_safety_keywords"]))
+    assert keywords & {"sfw", "nsfw"} == {"sfw"}
+
+
+def test_malformed_metadata_does_not_block_the_checkpoint(env):
+    result = SafetyService(env.sf).decide(env.assets["malformed_metadata"], "nsfw")
+    assert result["exif_verified"] is True
+
+
+# ── verification and the refreshed hash ───────────────────────────────────────
+
+
+def test_a_verified_checkpoint_refreshes_the_recorded_bytes(env):
+    """exiftool rewrites the container, so the stored SHA-256 must be the new one —
+    upload compares against exactly these bytes."""
+    asset_id = env.assets["no_exif"]
+    path = env.files["no_exif"]
+    SafetyService(env.sf).decide(asset_id, "sfw")
+
+    with env.sf() as session:
+        asset = session.get(Asset, asset_id)
+        row = session.get(ExifProjection, (asset_id, "safety"))
+    import hashlib
+
+    on_disk = hashlib.sha256(path.read_bytes()).hexdigest()
+    assert asset.current_sha256 == on_disk
+    assert row.state == "verified" and row.result_file_sha256 == on_disk
+    assert json.loads(row.desired_json) == {"add": ["sfw"], "remove": ["nsfw"]}
+
+
+def test_the_projection_survives_a_restart(env):
+    asset_id = env.assets["user_exif"]
+    SafetyService(env.sf).decide(asset_id, "nsfw")
+
+    engine = create_db_engine(env.config.database_url)  # a fresh connection, as a restart is
+    try:
+        with create_session_factory(engine)() as session:
+            row = session.get(ExifProjection, (asset_id, "safety"))
+            assert row.state == "verified" and row.verified_at is not None
+    finally:
+        engine.dispose()
+
+
+# ── divergence ────────────────────────────────────────────────────────────────
+
+
+def test_a_field_the_stage_does_not_own_changing_is_divergent(env, monkeypatch):
+    """Something rewrote the artist while the safety keyword was being written.
+
+    The stage must not call that verified, must record what moved, and must not put
+    the old value back — the file is now a question for a human.
+    """
+    asset_id = env.assets["user_exif"]
+    path = env.files["user_exif"]
+    real_apply = exiftool.apply_keywords
+
+    def sabotage(target, *, add=(), remove=()):
+        ok = real_apply(target, add=add, remove=remove)
+        subprocess.run(
+            ["exiftool", "-m", "-overwrite_original", "-Artist=Someone Else", str(target)],
+            capture_output=True,
+            check=False,
+        )
+        return ok
+
+    monkeypatch.setattr(exiftool, "apply_keywords", sabotage)
+    result = SafetyService(env.sf).decide(asset_id, "nsfw")
+
+    assert result["exif_verified"] is False, "a divergent checkpoint is not verified"
+    with env.sf() as session:
+        row = session.get(ExifProjection, (asset_id, "safety"))
+        asset = session.get(Asset, asset_id)
+    assert row.state == "divergent"
+    assert "EXIF:IFD0:Artist" in json.loads(row.divergent_fields)
+    assert row.verified_at is None
+    # Not repaired, and not silently accepted as the current verified bytes.
+    assert snapshot(path)["EXIF:IFD0:Artist"] == "Someone Else"
+    assert asset.current_sha256 is None
+
+
+def test_a_divergent_asset_shows_up_in_the_review_queue(env, monkeypatch):
+    asset_id = env.assets["user_exif"]
+    monkeypatch.setattr(
+        exif_checkpoint,
+        "run",
+        lambda *args, **kwargs: exif_checkpoint.CheckpointResult(
+            exif_checkpoint.DIVERGENT, changed_fields=("EXIF:IFD0:Artist",), sha256="abc"
+        ),
+    )
+    SafetyService(env.sf).decide(asset_id, "sfw")
+
+    rows = SafetyService(env.sf).review_queue()["items"]
+    row = next(item for item in rows if item["asset_id"] == asset_id)
+    assert row["exif_state"] == "divergent"
+    assert row["exif_verified"] is False
+
+
+def test_a_write_that_does_not_take_is_a_failure_not_a_verification(env, monkeypatch):
+    monkeypatch.setattr(exiftool, "apply_keywords", lambda *a, **k: False)
+    result = SafetyService(env.sf).decide(env.assets["no_exif"], "sfw")
+    assert result["exif_verified"] is False
+    assert exif_checkpoint.state_for(env.sf, env.assets["no_exif"], "safety") == "failed"
+
+
+def test_unreadable_metadata_is_a_failure_not_an_empty_snapshot(env, monkeypatch):
+    """``None`` from exiftool means "cannot answer"; treating it as "nothing there"
+    would make every field look preserved."""
+    monkeypatch.setattr(exiftool, "read_all", lambda path: None)
+    result = exif_checkpoint.run(str(env.files["no_exif"]), add=("sfw",))
+    assert result.state == "failed" and result.reason == "metadata_unreadable"
+
+
+def test_analysis_records_its_own_projection_separately(env):
+    asset_id = env.assets["user_exif"]
+    SafetyService(env.sf).decide(asset_id, "sfw")
+    AnalysisService(env.sf, provider=Provider(["pier"]), library_roots=(env.lib,)).run([asset_id])
+
+    assert exif_checkpoint.state_for(env.sf, asset_id, "safety") == "verified"
+    assert exif_checkpoint.state_for(env.sf, asset_id, "analysis") == "verified"
+
+
+# ── the comparison rules themselves ───────────────────────────────────────────
+
+
+def test_compare_ignores_owned_and_volatile_fields_only():
+    before = {
+        "EXIF:IFD0:Artist": "Ada",
+        "IPTC:Keywords": ["holiday"],
+        "File:System:FileSize": "3.8 kB",
+        "File:CurrentIPTCDigest": "aaa",
+        "XMP:XMP-x:XMPToolkit": "old",
+    }
+    after = {
+        "EXIF:IFD0:Artist": "Ada",
+        "IPTC:Keywords": ["holiday", "sfw"],
+        "File:System:FileSize": "3.9 kB",
+        "File:CurrentIPTCDigest": "bbb",
+        "XMP:XMP-x:XMPToolkit": "new",
+    }
+    assert exif_checkpoint.compare(before, after) == ()
+
+    after["EXIF:GPS:GPSLatitude"] = "48.1"  # an addition counts as much as a loss
+    del after["EXIF:IFD0:Artist"]
+    assert exif_checkpoint.compare(before, after) == ("EXIF:GPS:GPSLatitude", "EXIF:IFD0:Artist")

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

@@ -0,0 +1,335 @@
+"""US07-03: the golden media corpus, driven through the real decode path.
+
+Every case in ``tests/fixtures/media_corpus.py`` is exercised here. The claims:
+
+* a supported format/orientation/profile renders, with the orientation applied and
+  the decode bounded to roughly the size that was asked for;
+* a damaged, empty, lying, or gigapixel file becomes one precise item error — the
+  scan still finishes, the other assets still render, and a job that meets one keeps
+  running;
+* a failure leaves no temporary file behind, and startup removes only the temporaries
+  this service recognises;
+* cache invalidation follows the pixels, not the metadata.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import shutil
+import time
+import tracemalloc
+from types import SimpleNamespace
+
+import pytest
+from PIL import Image
+
+from photo_pipeline import imaging
+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, Thumbnail
+from photo_pipeline.services.inventory import InventoryService
+from photo_pipeline.services.thumbnails import (
+    TEMP_SUFFIX,
+    ThumbnailError,
+    ThumbnailService,
+)
+from tests.fixtures.media_corpus import CASES, CASES_BY_ID, build_corpus
+
+RENDERABLE = [case for case in CASES if case.expect_error is None]
+DAMAGED = [case for case in CASES if case.expect_error is not None]
+
+needs_exiftool = pytest.mark.skipif(
+    shutil.which("exiftool") is None, reason="exiftool not installed"
+)
+
+
+@pytest.fixture(scope="module")
+def corpus(tmp_path_factory):
+    root = tmp_path_factory.mktemp("corpus")
+    return SimpleNamespace(root=root, files=build_corpus(root))
+
+
+@pytest.fixture
+def env(tmp_path, corpus):
+    """A library holding the whole corpus, scanned into a fresh database."""
+    data = tmp_path / "data"
+    data.mkdir()
+    lib = tmp_path / "lib"
+    shutil.copytree(corpus.root, lib)
+    config = Config.from_env(
+        {"PHOTO_PIPELINE_DATA_DIR": str(data), "PHOTO_PIPELINE_LIBRARY_ROOTS": str(lib)}
+    )
+    run_migrations(config.database_url)
+    engine = create_db_engine(config.database_url)
+    sf = create_session_factory(engine)
+    scan = InventoryService(sf).scan(lib)
+    by_id = {
+        case.id: next(
+            (aid for path, aid in scan.asset_ids.items() if path.endswith(case.filename)), None
+        )
+        for case in CASES
+    }
+    yield SimpleNamespace(
+        config=config, lib=lib, sf=sf, scan=scan, assets=by_id,
+        thumbs=ThumbnailService(sf, config),
+    )
+    engine.dispose()
+
+
+# ── the manifest itself ───────────────────────────────────────────────────────
+
+
+def test_the_manifest_is_internally_consistent():
+    assert len({case.id for case in CASES}) == len(CASES), "logical ids must be unique"
+    assert len({case.filename for case in CASES}) == len(CASES), "paths must be unique"
+    for case in CASES:
+        assert case.kind in {"format", "orientation", "profile", "damaged", "metadata"}
+        assert case.expect_error in (None, "unsupported_image", "image_too_large")
+
+
+def test_the_corpus_regenerates_byte_for_byte(tmp_path):
+    """A golden corpus that drifts between runs cannot be a golden corpus."""
+    first = build_corpus(tmp_path / "one")
+    second = build_corpus(tmp_path / "two")
+    digests = {
+        case_id: (
+            hashlib.sha256(first[case_id].read_bytes()).hexdigest(),
+            hashlib.sha256(second[case_id].read_bytes()).hexdigest(),
+        )
+        for case_id in first
+    }
+    drifted = [case_id for case_id, (a, b) in digests.items() if a != b]
+    assert not drifted, f"non-deterministic fixtures: {drifted}"
+
+
+# ── bounded decode ────────────────────────────────────────────────────────────
+
+
+@pytest.mark.parametrize("case", RENDERABLE, ids=lambda case: case.id)
+def test_every_supported_case_renders(case, env):
+    asset_id = env.assets[case.id]
+    assert asset_id, f"{case.id} was not discovered by the scan"
+    path = env.thumbs.generate(asset_id, 256)
+    with Image.open(path) as thumb:
+        assert thumb.format == "WEBP"
+        assert max(thumb.size) <= 256
+        assert min(thumb.size) >= 1
+
+
+@pytest.mark.parametrize("orientation", range(1, 9))
+def test_exif_orientation_is_applied_before_resizing(orientation, env):
+    """All eight tags: the 90° ones must come out portrait from a landscape source."""
+    asset_id = env.assets[f"orientation_{orientation}"]
+    with Image.open(env.thumbs.generate(asset_id, 256)) as thumb:
+        rotated = orientation in (5, 6, 7, 8)
+        assert (thumb.height > thumb.width) is rotated
+
+
+def test_transparency_and_grayscale_survive_the_pipeline(env):
+    with Image.open(env.thumbs.generate(env.assets["png_alpha"], 256)) as thumb:
+        assert "A" in thumb.getbands()
+    with Image.open(env.thumbs.generate(env.assets["grayscale"], 256)) as thumb:
+        assert thumb.size == (100, 100)  # smaller than the request: never upscaled
+
+
+def test_a_broken_colour_profile_still_produces_a_picture(env):
+    """An unusable ICC profile is a metadata problem, not a reason to lose the preview."""
+    for case_id in ("icc_tagged", "broken_icc"):
+        with Image.open(env.thumbs.generate(env.assets[case_id], 256)) as thumb:
+            assert thumb.size[0] > 0
+
+
+def test_a_large_jpeg_is_not_decoded_at_full_resolution(env):
+    """12 megapixels would be ~36 MB of pixels; the draft decode keeps it far below."""
+    tracemalloc.start()
+    try:
+        env.thumbs.generate(env.assets["large_jpeg"], 256)
+        _, peak = tracemalloc.get_traced_memory()
+    finally:
+        tracemalloc.stop()
+    assert peak < 12_000_000, f"decode peaked at {peak} bytes"
+
+
+# ── damaged inputs ────────────────────────────────────────────────────────────
+
+
+@pytest.mark.parametrize("case", DAMAGED, ids=lambda case: case.id)
+def test_every_damaged_case_is_a_precise_item_error(case, env):
+    asset_id = env.assets[case.id]
+    assert asset_id, f"{case.id} was not discovered by the scan"
+    with pytest.raises(ThumbnailError) as raised:
+        env.thumbs.generate(asset_id, 256)
+    assert raised.value.code == case.expect_error
+    # Persisted, so a broken original is not re-decoded on every request...
+    with env.sf() as session:
+        rows = [
+            row
+            for row in session.query(Thumbnail).all()
+            if row.asset_id == asset_id and row.state == "error"
+        ]
+    assert rows and rows[0].error_code == case.expect_error
+    # ...and the cached failure is the same precise error, not a generic one.
+    with pytest.raises(ThumbnailError) as again:
+        env.thumbs.generate(asset_id, 256)
+    assert again.value.code == case.expect_error
+
+
+def test_a_gigapixel_header_is_refused_quickly_and_cheaply(env):
+    """The refusal must come from the declared size, not from decoding it."""
+    tracemalloc.start()
+    started = time.monotonic()
+    try:
+        with pytest.raises(ThumbnailError):
+            env.thumbs.generate(env.assets["bomb_header"], 1280)
+        _, peak = tracemalloc.get_traced_memory()
+    finally:
+        tracemalloc.stop()
+    assert time.monotonic() - started < 5
+    assert peak < 5_000_000, f"a 3.6 gigapixel header allocated {peak} bytes"
+
+
+def test_one_broken_file_stops_neither_the_scan_nor_its_neighbours(env):
+    """The whole corpus is in the library, damaged files included."""
+    assert len(env.scan.asset_ids) >= len(CASES) - 1  # the empty file has no pixels to hash
+    for case in RENDERABLE[:5]:
+        assert env.thumbs.generate(env.assets[case.id], 256).exists()
+
+
+def test_a_job_that_meets_a_broken_file_gets_evidence_not_an_exception(env):
+    """``ensure_protected`` is what the archive lane calls; a plan must not die on
+    one unreadable original."""
+    evidence = env.thumbs.ensure_protected(env.assets["corrupt_png"])
+    assert evidence["state"] == "unsupported"
+    assert evidence["error_code"] == "unsupported_image"
+    assert env.thumbs.ensure_protected(env.assets["jpeg"])["state"] == "ready"
+
+
+def test_undecodable_files_do_not_break_hashing(env):
+    """Discovery records what it can: byte identity always, pixel identity when the
+    file has pixels."""
+    with env.sf() as session:
+        assets = {a.current_path: a for a in session.query(Asset).all()}
+    for case in DAMAGED:
+        asset = next((a for path, a in assets.items() if path.endswith(case.filename)), None)
+        if asset is None:
+            continue
+        assert asset.current_sha256, "byte identity is always available"
+        assert asset.pixel_sha256 is None, "undecodable files must not invent pixel identity"
+
+
+# ── cache lifecycle ───────────────────────────────────────────────────────────
+
+
+def test_a_failed_render_leaves_no_temporary_behind(env):
+    with pytest.raises(ThumbnailError):
+        env.thumbs.generate(env.assets["truncated_jpeg"], 256)
+    leftovers = list(env.config.thumbnail_cache_dir.rglob(f"*{TEMP_SUFFIX}"))
+    assert leftovers == []
+
+
+def test_startup_cleanup_removes_only_recognised_temporaries(env):
+    cache = env.config.thumbnail_cache_dir
+    env.thumbs.generate(env.assets["jpeg"], 256)  # creates the cache directory
+    stale = cache / f".abc123{TEMP_SUFFIX}"
+    stale.write_bytes(b"half a thumbnail")
+    innocent = cache / "keep-me.webp"
+    innocent.write_bytes(b"not a temporary")
+
+    assert env.thumbs.cleanup_temp_files() == 1
+    assert not stale.exists()
+    assert innocent.exists()
+    assert list(cache.rglob("*.webp")), "real cache entries survive"
+
+
+def test_metadata_only_change_reuses_the_thumbnail_and_a_pixel_change_does_not(env):
+    """The cache key follows the pixels: an EXIF edit must not cost a re-render."""
+    asset_id = env.assets["jpeg"]
+    first = env.thumbs.generate(asset_id, 256)
+
+    with env.sf() as session:  # what a rescan records after an EXIF-only edit
+        asset = session.get(Asset, asset_id)
+        asset.current_sha256 = "different-bytes"
+        session.commit()
+    assert env.thumbs.generate(asset_id, 256) == first
+
+    with env.sf() as session:  # a genuine pixel change
+        asset = session.get(Asset, asset_id)
+        asset.pixel_sha256 = "different-pixels"
+        session.commit()
+    assert env.thumbs.generate(asset_id, 256) != first
+
+
+def test_a_deleted_cache_file_is_regenerated(env):
+    asset_id = env.assets["png"]
+    path = env.thumbs.generate(asset_id, 256)
+    path.unlink()
+    regenerated = env.thumbs.generate(asset_id, 256)
+    assert regenerated == path and regenerated.exists()
+
+
+# ── the imaging door itself ───────────────────────────────────────────────────
+
+
+def test_open_image_translates_every_decoder_failure(corpus):
+    """Callers must be able to catch two typed errors, never bare ``Exception``."""
+    for case in DAMAGED:
+        expected = imaging.ImageTooLarge if case.expect_error == "image_too_large" else imaging.UndecodableImage
+        with pytest.raises(expected):
+            with imaging.open_image(corpus.files[case.id]) as image:
+                image.load()
+
+
+def test_open_image_refuses_more_pixels_than_the_caller_allowed(corpus):
+    with pytest.raises(imaging.ImageTooLarge):
+        with imaging.open_image(corpus.files["jpeg"], max_pixels=100):
+            pass
+    with imaging.open_image(corpus.files["jpeg"], max_pixels=100_000) as image:
+        assert image.size == (320, 240)
+
+
+@needs_exiftool
+def test_malformed_metadata_does_not_stop_the_picture(env):
+    """A broken EXIF block is a metadata fact, not a decode failure."""
+    assert env.thumbs.generate(env.assets["malformed_metadata"], 256).exists()
+    with env.sf() as session:
+        asset = session.get(Asset, env.assets["malformed_metadata"])
+    assert asset.pixel_sha256, "pixels are still identifiable"
+
+
+def test_scoring_never_relaxes_truncated_image_handling(corpus):
+    """The donor's process-global ``LOAD_TRUNCATED_IMAGES`` is gone for good.
+
+    It is global state: switching it on for the safety model would also switch it on
+    for hashing and preview rendering in the same process, and half a file would
+    silently become a valid picture (donor_ledger: nt-score-model).
+    """
+    import inspect
+
+    from PIL import ImageFile
+
+    from photo_pipeline.integrations import nsfw_model
+
+    code = [
+        line
+        for line in inspect.getsource(nsfw_model).splitlines()
+        if not line.strip().startswith("#")
+    ]
+    assert not any("LOAD_TRUNCATED_IMAGES" in line for line in code)
+
+    # And the door holds even when something else in the process turned it on —
+    # the frozen donor does exactly that when the characterization suite imports it.
+    previous = ImageFile.LOAD_TRUNCATED_IMAGES
+    ImageFile.LOAD_TRUNCATED_IMAGES = True
+    try:
+        with pytest.raises(imaging.UndecodableImage):
+            with imaging.open_image(corpus.files["truncated_jpeg"]) as image:
+                image.load()
+        assert ImageFile.LOAD_TRUNCATED_IMAGES is True, "the caller's setting is restored"
+    finally:
+        ImageFile.LOAD_TRUNCATED_IMAGES = previous
+
+
+def test_every_manifest_case_is_exercised():
+    """The corpus lint: no fixture may sit in the manifest untested."""
+    covered = {case.id for case in RENDERABLE} | {case.id for case in DAMAGED}
+    assert covered == set(CASES_BY_ID)

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

@@ -137,6 +137,25 @@
     ],
     "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"
+    ],
+    "US07-03": [
+      "tests/integration/test_media_hardening.py",
+      "tests/integration/test_exif_checkpoints.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