US06-01: Configure and Preflight Archive Destinations

README.md

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

frontend/index.html

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

frontend/js/api.js

@@ -138,31 +138,4 @@ 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,5 +1,4 @@
 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";
@@ -367,7 +366,6 @@ 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"));
 }
@@ -376,6 +374,5 @@ function render() {
 setRender(render);
 setRenamesRender(render);
 setUploadsRender(render);
-setArchiveRender(render);
 onRouteChange(render);
 render();

frontend/js/archive.js

@@ -1,870 +0,0 @@
-// 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]}`;
-}

migrations/versions/0012_archive_plans.py

@@ -1,96 +0,0 @@
-"""Archive plans, per-item transfer journal, and archived asset location (US06-02).
-
-Revision ID: 0012_archive_plans
-Revises: 0011_archive_locations
-Create Date: 2026-08-16
-
-The journal is what makes removing an original recoverable: every item records its
-source, destination, expected hash, and the state it had reached before the process
-died. Assets gain the archive location and relative path so an offline original is
-still explained rather than looking missing.
-"""
-
-import sqlalchemy as sa
-from alembic import op
-
-revision = "0012_archive_plans"
-down_revision = "0011_archive_locations"
-branch_labels = None
-depends_on = None
-
-
-def upgrade() -> None:
-    op.create_table(
-        "archive_plans",
-        sa.Column("id", sa.String(), primary_key=True),
-        sa.Column(
-            "location_id",
-            sa.String(),
-            sa.ForeignKey("archive_locations.id"),
-            nullable=False,
-            index=True,
-        ),
-        # The preflight token this plan was approved against.
-        sa.Column("token", sa.String(), nullable=False),
-        sa.Column("albums", sa.String(), nullable=True),  # JSON array
-        # planned | applying | complete | failed
-        sa.Column("state", sa.String(), nullable=False, server_default="planned"),
-        sa.Column("schema_version", sa.Integer(), nullable=False, server_default="1"),
-        sa.Column("asset_count", sa.Integer(), nullable=False, server_default="0"),
-        sa.Column("byte_size", sa.Integer(), nullable=False, server_default="0"),
-        # Bumped on every claim and used as the fencing token.
-        sa.Column("version", sa.Integer(), nullable=False, server_default="1"),
-        sa.Column("worker_id", sa.String(), nullable=True),
-        sa.Column("completed_at", sa.DateTime(timezone=True), nullable=True),
-        sa.Column(
-            "created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()
-        ),
-        sa.Column(
-            "updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()
-        ),
-    )
-    op.create_table(
-        "archive_operations",
-        sa.Column("id", sa.String(), primary_key=True),
-        sa.Column(
-            "plan_id",
-            sa.String(),
-            sa.ForeignKey("archive_plans.id", ondelete="CASCADE"),
-            nullable=False,
-            index=True,
-        ),
-        sa.Column("sequence", sa.Integer(), nullable=False),
-        sa.Column("album", sa.String(), nullable=False),
-        sa.Column("asset_id", sa.String(), sa.ForeignKey("assets.id"), nullable=False, index=True),
-        sa.Column("source_path", sa.String(), nullable=False),
-        sa.Column("destination_path", sa.String(), nullable=False),
-        # Relative to the location root: the medium can be mounted anywhere later.
-        sa.Column("archive_path", sa.String(), nullable=False),
-        sa.Column("expected_sha256", sa.String(), nullable=False),
-        sa.Column("byte_size", sa.Integer(), nullable=True),
-        sa.Column("same_filesystem", sa.Boolean(), nullable=True),
-        # planned | transferring | verified | removing | complete | failed
-        sa.Column("journal_state", sa.String(), nullable=False, server_default="planned"),
-        sa.Column("attempt_count", sa.Integer(), nullable=False, server_default="0"),
-        sa.Column("fencing_token", sa.Integer(), nullable=True),
-        sa.Column("worker_id", sa.String(), nullable=True),
-        sa.Column("verified_at", sa.DateTime(timezone=True), nullable=True),
-        sa.Column("removed_at", sa.DateTime(timezone=True), nullable=True),
-        sa.Column("error_code", sa.String(), nullable=True),
-        sa.Column("error_message", sa.String(), nullable=True),
-        sa.Column(
-            "updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()
-        ),
-        sa.UniqueConstraint("plan_id", "sequence", name="uq_archive_operations_plan_sequence"),
-    )
-    # Plain columns: SQLite cannot ALTER a table to add a foreign key, and the
-    # relationship is enforced by the service that writes them.
-    op.add_column("assets", sa.Column("archive_location_id", sa.String(), nullable=True))
-    op.add_column("assets", sa.Column("archive_path", sa.String(), nullable=True))
-
-
-def downgrade() -> None:
-    op.drop_column("assets", "archive_path")
-    op.drop_column("assets", "archive_location_id")
-    op.drop_table("archive_operations")
-    op.drop_table("archive_plans")

migrations/versions/0013_protected_thumbnails.py

@@ -1,29 +0,0 @@
-"""Protected thumbnails (US06-03).
-
-Revision ID: 0013_protected_thumbnails
-Revises: 0012_archive_plans
-Create Date: 2026-08-16
-
-A protected thumbnail is the durable comparison preview of an asset whose
-original has left active storage. It is evidence rather than cache, so the LRU
-quota must not evict it: the archive medium may be offline when it is needed.
-"""
-
-import sqlalchemy as sa
-from alembic import op
-
-revision = "0013_protected_thumbnails"
-down_revision = "0012_archive_plans"
-branch_labels = None
-depends_on = None
-
-
-def upgrade() -> None:
-    op.add_column(
-        "thumbnails",
-        sa.Column("protected", sa.Boolean(), nullable=False, server_default=sa.false()),
-    )
-
-
-def downgrade() -> None:
-    op.drop_column("thumbnails", "protected")

migrations/versions/0014_restore_plans.py

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

photo_pipeline/api/routes/archives.py

@@ -1,10 +1,9 @@
-"""Archive location, preflight, plan, and restore API (US06-01, US06-02, US06-04).
+"""Archive location and preflight API (US06-01).
 
 Registering a location writes a marker onto the medium; preflight is a command
 rather than a read, because it probes the destination, hashes the scope, and issues
-the token an archive plan must present. Creating a plan writes only database rows —
+the token a later archive plan must present (US06-02). Neither endpoint moves or
-the transfer itself runs on the durable ``archive`` lane, never in the request
+removes a single library file.
-thread, because it removes originals.
 """
 
 from __future__ import annotations
@@ -13,18 +12,12 @@ from fastapi import APIRouter, Request
 from fastapi.responses import JSONResponse
 from pydantic import BaseModel
 
-from photo_pipeline.jobs.domain_handlers import ARCHIVE_LOCK, ARCHIVE_PLAN, RESTORE_PLAN
 from photo_pipeline.services.archives import ArchiveError, ArchiveService
-from photo_pipeline.services.archive_transfer import ArchiveTransferService
-from photo_pipeline.services.jobs import JobBlocked, JobService
-from photo_pipeline.services.restores import RestoreService
 
 router = APIRouter(tags=["archives"])
 
-# Which failures are the caller's request (422), a missing thing (404), or state
+# Which failures are the caller's request (422) and which are a missing thing (404).
-# that changed under the caller (409).
+NOT_FOUND_CODES = {"unknown_location"}
-NOT_FOUND_CODES = {"unknown_location", "unknown_plan"}
-CONFLICT_CODES = {"stale_token", "stale_plan", "archive_pending"}
 
 
 class RegisterLocationRequest(BaseModel):
@@ -38,42 +31,12 @@ class PreflightRequest(BaseModel):
     albums: list[str] | None = None
 
 
-class CreatePlanRequest(PreflightRequest):
-    # The token of the preflight the user approved; a stale one is refused.
-    token: str
-
-
-class RestoreRequest(BaseModel):
-    location_id: str
-    # ``None`` means every asset archived at this location.
-    asset_ids: list[str] | None = None
-
-
-class CreateRestoreRequest(RestoreRequest):
-    token: str
-
-
 def _service(request: Request) -> ArchiveService:
     return ArchiveService(request.app.state.session_factory, config=request.app.state.config)
 
 
-def _restores(request: Request) -> RestoreService:
-    return RestoreService(request.app.state.session_factory, config=request.app.state.config)
-
-
-def _transfers(request: Request) -> ArchiveTransferService:
-    return ArchiveTransferService(
-        request.app.state.session_factory, config=request.app.state.config
-    )
-
-
 def _error(error: ArchiveError) -> JSONResponse:
-    if error.code in NOT_FOUND_CODES:
+    status = 404 if error.code in NOT_FOUND_CODES else 422
-        status = 404
-    elif error.code in CONFLICT_CODES:
-        status = 409
-    else:
-        status = 422
     return JSONResponse(
         status_code=status, content={"error": {"code": error.code, "message": str(error)}}
     )
@@ -98,137 +61,3 @@ def preflight(body: PreflightRequest, request: Request):
         return _service(request).preflight(body.location_id, body.albums)
     except ArchiveError as error:
         return _error(error)
-
-
-@router.post("/archive-plans", status_code=201)
-def create_plan(body: CreatePlanRequest, request: Request):
-    """Turn an approved preflight into a durable, journaled plan. Nothing moves."""
-    try:
-        return _transfers(request).create(body.location_id, body.albums, token=body.token)
-    except ArchiveError as error:
-        return _error(error)
-
-
-@router.get("/archive-plans")
-def list_plans(request: Request) -> dict:
-    return {"plans": _transfers(request).list()}
-
-
-@router.get("/archive-plans/{plan_id}")
-def get_plan(plan_id: str, request: Request):
-    plan = _transfers(request).get(plan_id)
-    if plan is None:
-        return _error(ArchiveError("unknown_plan", f"unknown archive plan {plan_id}"))
-    return plan
-
-
-@router.post("/archive-plans/{plan_id}/apply")
-def apply_plan(plan_id: str, request: Request):
-    """Queue the transfer on the archiver lane. The worker removes the sources."""
-    service = _transfers(request)
-    plan = service.get(plan_id)
-    if plan is None:
-        return _error(ArchiveError("unknown_plan", f"unknown archive plan {plan_id}"))
-    if service.journal.blocks_mutation():
-        unresolved = [row for row in service.journal.incomplete() if row["plan_id"] != plan_id]
-        if unresolved:
-            return _error(
-                ArchiveError(
-                    "archive_pending",
-                    f"an unresolved archive operation ({unresolved[0]['id']}) must be recovered",
-                )
-            )
-    try:
-        job = JobService(request.app.state.session_factory).enqueue(
-            ARCHIVE_PLAN,
-            lock=ARCHIVE_LOCK,
-            # One queued attempt per plan version: a double-clicked apply reuses it.
-            idempotency_key=f"archive:{plan_id}:{plan['version']}",
-            items=[plan_id],
-        )
-    except JobBlocked as error:
-        return JSONResponse(
-            status_code=409, content={"error": {"code": error.code, "message": str(error)}}
-        )
-    return {"plan_id": plan_id, "job": job}
-
-
-@router.post("/restore-preflight")
-def restore_preflight(body: RestoreRequest, request: Request):
-    """Validate restoring archived assets back into the library. Nothing moves."""
-    try:
-        return _restores(request).preflight(body.location_id, body.asset_ids)
-    except ArchiveError as error:
-        return _error(error)
-
-
-@router.post("/restore-plans", status_code=201)
-def create_restore_plan(body: CreateRestoreRequest, request: Request):
-    try:
-        return _restores(request).create(body.location_id, body.asset_ids, token=body.token)
-    except ArchiveError as error:
-        return _error(error)
-
-
-@router.get("/restore-plans")
-def list_restore_plans(request: Request) -> dict:
-    return {"plans": _restores(request).list()}
-
-
-@router.get("/restore-plans/{plan_id}")
-def get_restore_plan(plan_id: str, request: Request):
-    plan = _restores(request).get(plan_id)
-    if plan is None:
-        return _error(ArchiveError("unknown_plan", f"unknown restore plan {plan_id}"))
-    return plan
-
-
-@router.post("/restore-plans/{plan_id}/apply")
-def apply_restore_plan(plan_id: str, request: Request):
-    """Queue the restore on the archiver lane — the same single lane as archiving,
-    because both move the same originals."""
-    service = _restores(request)
-    plan = service.get(plan_id)
-    if plan is None:
-        return _error(ArchiveError("unknown_plan", f"unknown restore plan {plan_id}"))
-    unresolved = [row for row in service.journal.incomplete() if row["plan_id"] != plan_id]
-    if unresolved:
-        return _error(
-            ArchiveError(
-                "archive_pending",
-                f"an unresolved archive operation ({unresolved[0]['id']}) must be recovered",
-            )
-        )
-    try:
-        job = JobService(request.app.state.session_factory).enqueue(
-            RESTORE_PLAN,
-            lock=ARCHIVE_LOCK,
-            idempotency_key=f"restore:{plan_id}:{plan['version']}",
-            items=[plan_id],
-        )
-    except JobBlocked as error:
-        return JSONResponse(
-            status_code=409, content={"error": {"code": error.code, "message": str(error)}}
-        )
-    return {"plan_id": plan_id, "job": job}
-
-
-@router.get("/restore-recovery")
-def restore_recovery_status(request: Request) -> dict:
-    return _restores(request).recovery_status()
-
-
-@router.post("/restore-recovery/resolve")
-def resolve_restore_recovery(request: Request) -> dict:
-    return _restores(request).recover()
-
-
-@router.get("/archive-recovery")
-def recovery_status(request: Request) -> dict:
-    """What an interrupted transfer left behind, straight from journal + disk."""
-    return _transfers(request).recovery_status()
-
-
-@router.post("/archive-recovery/resolve")
-def resolve_recovery(request: Request) -> dict:
-    return _transfers(request).recover()

photo_pipeline/jobs/domain_handlers.py

@@ -1,12 +1,10 @@
-"""Domain job handlers: safety scoring, content analysis, uploads, archive
+"""Domain job handlers: safety scoring, content analysis, uploads (US02-06, US05-02).
-transfers, restores (US02-06, US05-02, US06-02, US06-04).
 
-Importing this module registers the ``safety_score``, ``analysis``,
+Importing this module registers the ``safety_score``, ``analysis``, and
-``upload_batch``, ``archive_plan``, and ``restore_plan`` job types so the generic
+``upload_batch`` job types so the generic worker can run them per item. Each handler
-worker can run them per item. Each handler delegates to its service, which owns the real work and the
+delegates to its service, which owns the real work and the privacy gate. Handlers
-privacy gate. Handlers are idempotent: re-scoring or re-analyzing one asset is safe
+are idempotent: re-scoring or re-analyzing one asset is safe after an interrupted
-after an interrupted attempt, an upload batch refuses to re-run an attempt whose
+attempt, and an upload batch refuses to re-run an attempt whose outcome is unknown.
-outcome is unknown, and an archive plan skips items it already completed.
 
 Providers/models are the service defaults here (real NsfwModel / vision provider);
 tests exercise the services directly with injected fakes rather than the worker.
@@ -19,13 +17,12 @@ from photo_pipeline.jobs.handlers import Cancelled, JobContext, register
 SAFETY_SCORE = "safety_score"
 ANALYSIS = "analysis"
 UPLOAD_BATCH = "upload_batch"
-ARCHIVE_PLAN = "archive_plan"
-RESTORE_PLAN = "restore_plan"
 # Both mutate the library's metadata/derived state; one at a time (concept §one job).
 LIBRARY_WRITE_LOCK = "library_write"
 # The uploader lane: one album batch at a time (concept §16).
 UPLOAD_LOCK = "upload"
-# The archiver lane: one archive/restore plan at a time (concept §16).
+# The archiver lane: one archive/restore plan at a time (concept §16). No handler
+# runs on it yet (US06-02); preflight already refuses to plan around a held lease.
 ARCHIVE_LOCK = "archive"
 
 
@@ -57,36 +54,6 @@ def _upload_batch_item(batch_id: str, ctx: JobContext) -> None:
         raise RuntimeError(f"upload batch {batch_id} is {batch['state']}: {batch['error_code']}")
 
 
-def _archive_plan_item(plan_id: str, ctx: JobContext) -> None:
-    """One item = one archive plan. Item-level failures stay in the journal (the
-    source is then still there); only an unusable plan fails the job."""
-    from photo_pipeline.config import Config
-    from photo_pipeline.services.archive_transfer import ArchiveTransferService
-
-    config = ctx.config if ctx.config is not None else Config.from_env()
-    result = ArchiveTransferService(ctx.session_factory, config=config).apply(
-        plan_id, worker_id=ctx.worker_id
-    )
-    if result["failed"]:
-        raise RuntimeError(f"archive plan {plan_id}: {result['failed']} item(s) failed")
-
-
-def _restore_plan_item(plan_id: str, ctx: JobContext) -> None:
-    """One item = one restore plan. A restore removes nothing, so an item failure
-    simply leaves that asset archived (US06-04)."""
-    from photo_pipeline.config import Config
-    from photo_pipeline.services.restores import RestoreService
-
-    config = ctx.config if ctx.config is not None else Config.from_env()
-    result = RestoreService(ctx.session_factory, config=config).apply(
-        plan_id, worker_id=ctx.worker_id
-    )
-    if result["failed"]:
-        raise RuntimeError(f"restore plan {plan_id}: {result['failed']} item(s) failed")
-
-
 register(SAFETY_SCORE, _safety_score_item)
 register(ANALYSIS, _analysis_item)
 register(UPLOAD_BATCH, _upload_batch_item)
-register(ARCHIVE_PLAN, _archive_plan_item)
-register(RESTORE_PLAN, _restore_plan_item)

photo_pipeline/models/__init__.py

@@ -5,7 +5,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.archives import ArchiveLocation
 from photo_pipeline.models.assets import Asset, AssetPath
 from photo_pipeline.models.duplicates import (
     DuplicateCluster,
@@ -21,8 +21,6 @@ from photo_pipeline.models.workflow import AnalysisResult, SafetyReview
 __all__ = [
     "AlbumProposal",
     "ArchiveLocation",
-    "ArchiveOperation",
-    "ArchivePlan",
     "Asset",
     "AssetPath",
     "DuplicateCluster",

photo_pipeline/models/archives.py

@@ -1,4 +1,4 @@
-"""Archive location, plan, and transfer-journal persistence (US06-01, US06-02).
+"""Archive location persistence (US06-01).
 
 An archive location is a *medium*, not a path. External disks get mounted at
 different mountpoints, and a different disk can be mounted at the same one, so a
@@ -9,27 +9,13 @@ location therefore owns a marker file written onto the medium itself; its
 ``capabilities`` and ``state`` are the last probe result, kept so the UI can list
 locations without touching a sleeping disk. Preflight always re-probes — a stored
 state is a hint, never evidence.
-
-An ``ArchivePlan`` is one approved preflight turned into durable work, and each
-``ArchiveOperation`` is one file's crash-safe journal row (US06-02). The row records
-what the transfer *intends* to do before it does it — source, destination, expected
-hash — because after a crash that intent plus the files on disk is the only evidence
-available for deciding whether an original may be removed.
 """
 
 from __future__ import annotations
 
 from datetime import datetime
 
-from sqlalchemy import (
+from sqlalchemy import DateTime, String, func
-    Boolean,
-    DateTime,
-    ForeignKey,
-    Integer,
-    String,
-    UniqueConstraint,
-    func,
-)
 from sqlalchemy.orm import Mapped, mapped_column
 
 from photo_pipeline.db import Base
@@ -54,74 +40,3 @@ class ArchiveLocation(Base):
     updated_at: Mapped[datetime] = mapped_column(
         DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now()
     )
-
-
-class ArchivePlan(Base):
-    __tablename__ = "archive_plans"
-
-    id: Mapped[str] = mapped_column(String, primary_key=True)
-    location_id: Mapped[str] = mapped_column(
-        ForeignKey("archive_locations.id"), nullable=False, index=True
-    )
-    # The preflight token this plan was approved against; re-verified before apply.
-    token: Mapped[str] = mapped_column(String, nullable=False)
-    albums: Mapped[str | None] = mapped_column(String)  # JSON array
-    # archive | restore — the same journal read in the opposite direction (US06-04).
-    direction: Mapped[str] = mapped_column(String, nullable=False, default="archive")
-
-    # planned | applying | complete | failed
-    state: Mapped[str] = mapped_column(String, nullable=False, default="planned")
-    schema_version: Mapped[int] = mapped_column(Integer, nullable=False, default=1)
-    asset_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
-    byte_size: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
-    # Bumped on every claim and used as the fencing token, so a superseded attempt
-    # cannot commit.
-    version: Mapped[int] = mapped_column(Integer, nullable=False, default=1)
-    worker_id: Mapped[str | None] = mapped_column(String)
-    completed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
-
-    created_at: Mapped[datetime] = mapped_column(
-        DateTime(timezone=True), nullable=False, server_default=func.now()
-    )
-    updated_at: Mapped[datetime] = mapped_column(
-        DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now()
-    )
-
-
-class ArchiveOperation(Base):
-    __tablename__ = "archive_operations"
-    __table_args__ = (
-        UniqueConstraint("plan_id", "sequence", name="uq_archive_operations_plan_sequence"),
-    )
-
-    id: Mapped[str] = mapped_column(String, primary_key=True)
-    plan_id: Mapped[str] = mapped_column(
-        ForeignKey("archive_plans.id", ondelete="CASCADE"), nullable=False, index=True
-    )
-    sequence: Mapped[int] = mapped_column(Integer, nullable=False)
-    album: Mapped[str] = mapped_column(String, nullable=False)
-    asset_id: Mapped[str] = mapped_column(ForeignKey("assets.id"), nullable=False, index=True)
-
-    # archive: library → medium. restore: medium → library (US06-04). ``source_path``
-    # and ``destination_path`` always mean "from" and "to" for this direction.
-    direction: Mapped[str] = mapped_column(String, nullable=False, default="archive")
-    source_path: Mapped[str] = mapped_column(String, nullable=False)
-    destination_path: Mapped[str] = mapped_column(String, nullable=False)
-    # Relative to the location root, because the medium can be mounted elsewhere.
-    archive_path: Mapped[str] = mapped_column(String, nullable=False)
-    expected_sha256: Mapped[str] = mapped_column(String, nullable=False)
-    byte_size: Mapped[int | None] = mapped_column(Integer)
-    same_filesystem: Mapped[bool | None] = mapped_column(Boolean)
-
-    # planned | transferring | verified | removing | complete | failed
-    journal_state: Mapped[str] = mapped_column(String, nullable=False, default="planned")
-    attempt_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
-    fencing_token: Mapped[int | None] = mapped_column(Integer)
-    worker_id: Mapped[str | None] = mapped_column(String)
-    verified_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
-    removed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
-    error_code: Mapped[str | None] = mapped_column(String)
-    error_message: Mapped[str | None] = mapped_column(String)
-    updated_at: Mapped[datetime] = mapped_column(
-        DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now()
-    )

photo_pipeline/models/assets.py

@@ -35,18 +35,7 @@ class Asset(Base):
 
     discovered_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
     missing_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
-    # active | archiving | archived_online | archived_offline | restoring |
-    # missing_unexpected (concept §9). ``current_path`` is NULL once archived; the
-    # original is then explained by the location plus its relative archive path.
     availability_state: Mapped[str] = mapped_column(String, nullable=False, default="active")
-    # Not a declared foreign key: SQLite cannot add one to an existing table, so the
-    # link is written and read by the archive service (US06-02).
-    archive_location_id: Mapped[str | None] = mapped_column(String)
-    archive_path: Mapped[str | None] = mapped_column(String)
-    # Set when the archived copy was proven to hold bytes other than the recorded
-    # ones (US06-04). Restore refuses such an asset instead of accepting a different
-    # file; cleared as soon as a verification matches again.
-    archive_divergent_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
     # Duplicate canonical link: NULL when the asset is itself canonical or undecided.
     canonical_asset_id: Mapped[str | None] = mapped_column(ForeignKey("assets.id"))
 

photo_pipeline/models/thumbnails.py

@@ -11,7 +11,7 @@ from __future__ import annotations
 
 from datetime import datetime
 
-from sqlalchemy import Boolean, DateTime, ForeignKey, Integer, String, func
+from sqlalchemy import DateTime, ForeignKey, Integer, String, func
 from sqlalchemy.orm import Mapped, mapped_column
 
 from photo_pipeline.db import Base
@@ -29,9 +29,6 @@ class Thumbnail(Base):
     width: Mapped[int | None] = mapped_column(Integer)
     height: Mapped[int | None] = mapped_column(Integer)
     format: Mapped[str | None] = mapped_column(String)
-    # Durable comparison evidence for an archived asset: never evicted by the LRU
-    # quota, because the original may be on a medium that is no longer reachable.
-    protected: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
     created_at: Mapped[datetime] = mapped_column(
         DateTime(timezone=True), nullable=False, server_default=func.now()
     )

photo_pipeline/services/archive_journal.py

@@ -1,359 +0,0 @@
-"""Archive transfer journal — the durable record of every per-file transition
-(US06-02).
-
-Archiving is the only stage that deletes an original, so the journal exists to make
-one question answerable after any crash: *may this source file be removed?* Intent
-is written before the mutation it describes, and the recorded state plus the real
-files on disk are the sole basis for answering it later. This module owns the state
-machine, the durable writes, and the evidence table; it never touches a photo
-(:mod:`photo_pipeline.services.archive_transfer` does).
-
-Per-item state machine (concept §9 "Transfer and removal semantics"):
-
-```
-planned → transferring → verified → removing → complete
-              ↘             ↘          ↘ failed
-```
-
-A restore (US06-04) uses the same rows with ``direction='restore'``: it copies from
-the medium back into the library and removes nothing, so it goes ``verified →
-complete`` directly. ``source_path``/``destination_path`` always mean "from"/"to",
-which is why the evidence table below needs no direction of its own.
-
-- ``transferring`` — intent recorded; a temporary copy may exist, the destination
-  may or may not have been published. Nothing has been removed.
-- ``verified``     — the archived bytes exist at their final path, hash exactly as
-  recorded, and the manifest entry is durable. Only from here may a source go.
-- ``removing``     — the source removal is committed to; the source may already be
-  gone while the database still points at it.
-- ``complete``     — source absent, database updated, availability recorded.
-
-``classify`` labels each incomplete item from the journal plus disk evidence:
-
-- ``resumable`` — nothing was published; the source is intact, so applying again is
-  safe.
-- ``forward``   — the archived copy exists and matches its recorded hash, so the
-  remaining steps (manifest, removal, bookkeeping) can be finished deterministically.
-- ``manual``    — the evidence contradicts the journal (missing archive copy, wrong
-  bytes, source and archive both gone). Nothing is guessed and nothing is removed;
-  the item blocks unrelated mutations until a human decides.
-"""
-
-from __future__ import annotations
-
-from datetime import datetime, timezone
-from pathlib import Path
-
-from sqlalchemy import select
-from sqlalchemy.orm import sessionmaker
-
-from photo_pipeline.models import ArchiveOperation, ArchivePlan
-from photo_pipeline.services.hashing import sha256_file
-
-
-class ArchiveState:
-    PLANNED = "planned"
-    TRANSFERRING = "transferring"
-    VERIFIED = "verified"
-    REMOVING = "removing"
-    COMPLETE = "complete"
-    FAILED = "failed"
-
-
-ALLOWED_TRANSITIONS = {
-    ArchiveState.PLANNED: {ArchiveState.TRANSFERRING, ArchiveState.FAILED},
-    # From `transferring` the outcome is unknown until evidence is gathered, so it
-    # may resolve forward, back to planned (proven nothing was published), or fail.
-    ArchiveState.TRANSFERRING: {
-        ArchiveState.VERIFIED,
-        ArchiveState.PLANNED,
-        ArchiveState.FAILED,
-    },
-    ArchiveState.VERIFIED: {ArchiveState.REMOVING, ArchiveState.FAILED},
-    # No path back: once the source may be gone, only finishing is safe.
-    ArchiveState.REMOVING: {ArchiveState.COMPLETE, ArchiveState.FAILED},
-    ArchiveState.COMPLETE: set(),
-    # A retry re-enters `transferring`, which rechecks every precondition from
-    # scratch; recovery may also reset a failed item to `planned`.
-    ArchiveState.FAILED: {ArchiveState.PLANNED, ArchiveState.TRANSFERRING},
-}
-
-# A restore removes nothing, so it has no ``removing`` step: a verified published
-# copy is the whole job (US06-04). Keeping this as a separate table means the
-# archive direction still cannot reach ``complete`` without going through removal.
-RESTORE_TRANSITIONS = {
-    **ALLOWED_TRANSITIONS,
-    ArchiveState.VERIFIED: {ArchiveState.COMPLETE, ArchiveState.FAILED},
-}
-
-TERMINAL_STATES = frozenset({ArchiveState.COMPLETE})
-# States where this item may already have touched the filesystem.
-UNSAFE_STATES = frozenset({ArchiveState.TRANSFERRING, ArchiveState.VERIFIED, ArchiveState.REMOVING})
-
-# Which way the bytes move. Same rows, same evidence table, opposite direction.
-ARCHIVE = "archive"
-RESTORE = "restore"
-
-RESUMABLE = "resumable"
-FORWARD = "forward"
-MANUAL = "manual"
-
-
-class JournalError(RuntimeError):
-    pass
-
-
-class InvalidTransition(JournalError):
-    pass
-
-
-class JournalConflict(JournalError):
-    """Fencing check failed; a newer owner has taken over this operation."""
-
-
-def can_transition(current: str, target: str, direction: str = ARCHIVE) -> bool:
-    table = RESTORE_TRANSITIONS if direction == RESTORE else ALLOWED_TRANSITIONS
-    return target in table.get(current, set())
-
-
-def _now() -> datetime:
-    return datetime.now(timezone.utc)
-
-
-class ArchiveJournal:
-    def __init__(self, session_factory: sessionmaker) -> None:
-        self._session_factory = session_factory
-
-    # ── intent ────────────────────────────────────────────────────────────────
-
-    def begin(self, operation_id: str, *, worker_id: str, fencing_token: int) -> dict:
-        """Record the intent to transfer **before** touching the filesystem."""
-        with self._session_factory() as session:
-            row = self._require(session, operation_id)
-            if row.fencing_token is not None and fencing_token < row.fencing_token:
-                raise JournalConflict(
-                    f"stale fencing token {fencing_token} (current {row.fencing_token})"
-                )
-            if row.journal_state in TERMINAL_STATES:
-                raise InvalidTransition(f"{row.journal_state} is terminal")
-            if row.journal_state != ArchiveState.TRANSFERRING and not can_transition(
-                row.journal_state, ArchiveState.TRANSFERRING, row.direction
-            ):
-                raise InvalidTransition(f"{row.journal_state} -> {ArchiveState.TRANSFERRING}")
-            if row.journal_state != ArchiveState.TRANSFERRING:
-                row.attempt_count += 1
-            row.journal_state = ArchiveState.TRANSFERRING
-            row.worker_id = worker_id
-            row.fencing_token = fencing_token
-            row.error_code = row.error_message = None
-            row.updated_at = _now()
-            session.commit()
-            return _operation_dict(row)
-
-    # ── transitions ───────────────────────────────────────────────────────────
-
-    def transition(
-        self,
-        operation_id: str,
-        target: str,
-        *,
-        fencing_token: int | None = None,
-        error: tuple[str, str] | None = None,
-        same_filesystem: bool | None = None,
-    ) -> dict:
-        """Move one operation to ``target``, enforcing the state machine.
-
-        Re-entering the state an operation already holds is a no-op, which is what
-        makes recovery idempotent across repeated restarts.
-        """
-        with self._session_factory() as session:
-            row = self._require(session, operation_id)
-            if fencing_token is not None and row.fencing_token is not None:
-                if fencing_token < row.fencing_token:
-                    raise JournalConflict(
-                        f"stale fencing token {fencing_token} (current {row.fencing_token})"
-                    )
-            if same_filesystem is not None:
-                row.same_filesystem = same_filesystem
-            if row.journal_state == target:
-                session.commit()
-                return _operation_dict(row)  # idempotent
-            if not can_transition(row.journal_state, target, row.direction):
-                raise InvalidTransition(f"{row.journal_state} -> {target}")
-
-            row.journal_state = target
-            row.updated_at = _now()
-            if target == ArchiveState.VERIFIED:
-                row.verified_at = _now()
-            if target == ArchiveState.COMPLETE:
-                row.removed_at = _now()
-            if error:
-                row.error_code, row.error_message = error[0], error[1][:500]
-            elif target != ArchiveState.FAILED:
-                row.error_code = row.error_message = None
-            session.commit()
-            return _operation_dict(row)
-
-    # ── reads ─────────────────────────────────────────────────────────────────
-
-    def get(self, operation_id: str) -> dict | None:
-        with self._session_factory() as session:
-            row = session.get(ArchiveOperation, operation_id)
-            return _operation_dict(row) if row else None
-
-    def operations(self, plan_id: str) -> list[dict]:
-        with self._session_factory() as session:
-            rows = session.scalars(
-                select(ArchiveOperation)
-                .where(ArchiveOperation.plan_id == plan_id)
-                .order_by(ArchiveOperation.sequence)
-            )
-            return [_operation_dict(row) for row in rows]
-
-    def incomplete(self, *, direction: str | None = None) -> list[dict]:
-        """Every operation left in a non-terminal, non-planned state — the work a
-        restart has to reason about. Without ``direction`` this spans archives and
-        restores, because either one half-done blocks the other."""
-        with self._session_factory() as session:
-            stmt = select(ArchiveOperation).where(
-                ArchiveOperation.journal_state.not_in([*TERMINAL_STATES, ArchiveState.PLANNED])
-            )
-            if direction is not None:
-                stmt = stmt.where(ArchiveOperation.direction == direction)
-            rows = session.scalars(
-                stmt.order_by(ArchiveOperation.plan_id, ArchiveOperation.sequence)
-            )
-            return [_operation_dict(row) for row in rows]
-
-    # ── startup classification ────────────────────────────────────────────────
-
-    def classify(self, operation_id: str) -> dict:
-        """Classify one incomplete operation from the journal plus disk evidence.
-
-        Hashes the archived copy when one exists: "a file is at the destination" is
-        not evidence that the *right* bytes are, and only the right bytes justify
-        removing an original. Never mutates anything.
-        """
-        row = self.get(operation_id)
-        if row is None:
-            raise JournalError(f"unknown archive operation {operation_id!r}")
-        source = Path(row["source_path"])
-        destination = Path(row["destination_path"])
-        source_exists = source.exists()
-        destination_exists = destination.exists()
-        destination_matches = (
-            destination_exists and sha256_file(destination) == row["expected_sha256"]
-        )
-
-        classification, reason = _classify(
-            row["journal_state"], source_exists, destination_exists, destination_matches
-        )
-        return {
-            "operation_id": operation_id,
-            "plan_id": row["plan_id"],
-            "direction": row["direction"],
-            "album": row["album"],
-            "asset_id": row["asset_id"],
-            "source_path": row["source_path"],
-            "destination_path": row["destination_path"],
-            "journal_state": row["journal_state"],
-            "classification": classification,
-            "reason": reason,
-            "source_exists": source_exists,
-            "destination_exists": destination_exists,
-            "destination_matches": destination_matches,
-        }
-
-    def classify_all(self, *, direction: str | None = None) -> list[dict]:
-        return [self.classify(row["id"]) for row in self.incomplete(direction=direction)]
-
-    def blocks_mutation(self) -> bool:
-        """True when any item may have the library half-archived."""
-        return any(row["journal_state"] in UNSAFE_STATES for row in self.incomplete())
-
-    # ── plan-level ────────────────────────────────────────────────────────────
-
-    def plan_state(self, plan_id: str) -> str:
-        """Derive the plan's state from its items, so the summary can never disagree
-        with the journal."""
-        states = {row["journal_state"] for row in self.operations(plan_id)}
-        if not states:
-            return "planned"
-        if states <= {ArchiveState.COMPLETE}:
-            return "complete"
-        if states & {ArchiveState.FAILED}:
-            return "failed"
-        if states & UNSAFE_STATES:
-            return "applying"
-        return "planned"
-
-    def sync_plan_state(self, plan_id: str) -> str:
-        state = self.plan_state(plan_id)
-        with self._session_factory() as session:
-            plan = session.get(ArchivePlan, plan_id)
-            if plan is None:
-                raise JournalError(f"unknown archive plan {plan_id!r}")
-            if plan.state != state:
-                plan.state = state
-                plan.version += 1
-                plan.updated_at = _now()
-                if state == "complete" and plan.completed_at is None:
-                    plan.completed_at = _now()
-                session.commit()
-        return state
-
-    @staticmethod
-    def _require(session, operation_id: str) -> ArchiveOperation:
-        row = session.get(ArchiveOperation, operation_id)
-        if row is None:
-            raise JournalError(f"unknown archive operation {operation_id!r}")
-        return row
-
-
-def _classify(
-    state: str, source_exists: bool, destination_exists: bool, destination_matches: bool
-) -> tuple[str, str]:
-    """The evidence table. Kept a pure function so every combination is testable."""
-    if destination_exists and not destination_matches and state != ArchiveState.PLANNED:
-        # Someone else's file, or a partial/edited copy: never overwrite it, and
-        # never treat it as the durable archive that justifies a deletion.
-        return MANUAL, "the archived path holds bytes that are not the recorded ones"
-
-    if state in (ArchiveState.TRANSFERRING, ArchiveState.FAILED):
-        if destination_matches:
-            return FORWARD, "the archived copy is durable; finish the remaining steps"
-        if source_exists:
-            return RESUMABLE, "nothing was published; the source is intact"
-        return MANUAL, "neither the source nor a verified archive copy is present"
-
-    if state in (ArchiveState.VERIFIED, ArchiveState.REMOVING):
-        if destination_matches:
-            return FORWARD, "the archived copy is durable; finish the remaining steps"
-        return MANUAL, f"journal says {state} but the archived copy is missing"
-
-    return MANUAL, f"unhandled journal state {state}"
-
-
-def _operation_dict(row: ArchiveOperation) -> dict:
-    return {
-        "id": row.id,
-        "plan_id": row.plan_id,
-        "direction": row.direction,
-        "sequence": row.sequence,
-        "album": row.album,
-        "asset_id": row.asset_id,
-        "source_path": row.source_path,
-        "destination_path": row.destination_path,
-        "archive_path": row.archive_path,
-        "expected_sha256": row.expected_sha256,
-        "byte_size": row.byte_size,
-        "same_filesystem": row.same_filesystem,
-        "journal_state": row.journal_state,
-        "attempt_count": row.attempt_count,
-        "fencing_token": row.fencing_token,
-        "worker_id": row.worker_id,
-        "verified_at": row.verified_at.isoformat() if row.verified_at else None,
-        "removed_at": row.removed_at.isoformat() if row.removed_at else None,
-        "error_code": row.error_code,
-        "error_message": row.error_message,
-    }

photo_pipeline/services/archive_transfer.py

@@ -1,644 +0,0 @@
-"""Transfer an approved archive plan, verify it, and remove the active sources
-(US06-02).
-
-This is the only module that deletes originals from the photo library, so every
-step exists to make one promise keepable: **a source is removed only after the
-archived bytes are durable and proven identical.** The journal (US06-02,
-:mod:`photo_pipeline.services.archive_journal`) records intent before each mutation;
-this module performs the mutations and the recovery that reads that intent back.
-
-Per file the sequence is:
-
-```
-journal.begin (transferring)   ← intent persisted BEFORE any disk change
-recheck preconditions          ← source hash, free destination, no symlink
-copy to a temporary file       ← same directory, so the publish is atomic
-fsync, close, hash it back     ← read from disk; the write is not the evidence
-atomically publish             ← rename onto the final archive path
-append the manifest entry      ← durable on the medium itself, fsynced
-journal → verified
-journal → removing             ← intent to delete, persisted first
-re-verify the archive copy, remove the source, verify its absence
-current_path=NULL, close the path occurrence, availability + location recorded
-journal → complete
-```
-
-Same-filesystem albums may skip the copy and use an atomic ``rename`` instead
-(concept §9), but only when the shared device is proven at run time — never from the
-plan's stored guess — and the published file is hashed afterwards exactly as in the
-copy path.
-
-Rules that are never relaxed:
-
-- An occupied destination is never overwritten; the item fails with the source
-  untouched.
-- A source whose bytes no longer match the plan is never archived and never removed.
-- A crash resolves from journal + disk evidence only: an archive copy that is
-  missing or hashes differently blocks the item for a human instead of being
-  retried or, worse, treated as a successful archive.
-- Recovery is idempotent — repeated passes converge on the same state.
-"""
-
-from __future__ import annotations
-
-import json
-import os
-import shutil
-import uuid
-from datetime import datetime, timezone
-from pathlib import Path
-
-from sqlalchemy import select
-from sqlalchemy.orm import sessionmaker
-
-from photo_pipeline.config import Config
-from photo_pipeline.models import ArchiveLocation, ArchiveOperation, ArchivePlan, Asset, AssetPath
-from photo_pipeline.services.archive_journal import (
-    ARCHIVE,
-    MANUAL,
-    RESUMABLE,
-    ArchiveJournal,
-    ArchiveState,
-)
-from photo_pipeline.services.archives import MARKER_NAME, ArchiveError, ArchiveService
-from photo_pipeline.services.duplicates import DuplicateService
-from photo_pipeline.services.hashing import sha256_file
-from photo_pipeline.services.rename_apply import PreconditionFailed, maybe_fault
-from photo_pipeline.services.thumbnails import ThumbnailService
-
-# The per-medium manifest: one JSON line per archived file, appended and fsynced
-# before its source is removed. It lives with the bytes so the archive can still be
-# read back if the database is lost.
-MANIFEST_NAME = "archive-manifest.jsonl"
-MANIFEST_VERSION = 1
-TEMP_SUFFIX = ".part"
-TEMP_PREFIX = ".archive-"
-# Fault barrier between "the source is gone" and "the database knows it" — not a
-# journal state, but the transition crash tests care about most.
-SOURCE_REMOVED = "source_removed"
-
-APPLYABLE_PLAN_STATES = frozenset({"planned", "applying", "failed", "complete"})
-
-
-def _now() -> datetime:
-    return datetime.now(timezone.utc)
-
-
-class ArchiveTransferService:
-    def __init__(self, session_factory: sessionmaker, *, config: Config) -> None:
-        self._session_factory = session_factory
-        self._config = config
-        self.journal = ArchiveJournal(session_factory)
-
-    # ── plans ─────────────────────────────────────────────────────────────────
-
-    def create(self, location_id: str, albums: list[str] | None = None, *, token: str) -> dict:
-        """Turn an approved preflight into a durable plan.
-
-        The token is re-derived from a fresh preflight, so a plan can only be
-        created for the exact scope, bytes, and destination the user approved.
-        """
-        preflight = ArchiveService(self._session_factory, config=self._config).preflight(
-            location_id, albums
-        )
-        if not token or token != preflight["token"]:
-            raise ArchiveError("stale_token", "the archive preflight changed since it was approved")
-        if preflight["state"] != "ready":
-            codes = ", ".join(sorted({issue["code"] for issue in preflight["blockers"]})) or "-"
-            raise ArchiveError("blocked", f"the archive scope is blocked: {codes}")
-
-        root = Path(preflight["location"]["root"])
-        plan_id = str(uuid.uuid4())
-        with self._session_factory() as session:
-            session.add(
-                ArchivePlan(
-                    id=plan_id,
-                    location_id=location_id,
-                    token=token,
-                    albums=json.dumps(albums) if albums is not None else None,
-                    direction=ARCHIVE,
-                    state="planned",
-                    schema_version=MANIFEST_VERSION,
-                    asset_count=preflight["totals"]["assets"],
-                    byte_size=preflight["totals"]["bytes"],
-                )
-            )
-            session.flush()  # the plan row must exist before its items reference it
-            sequence = 0
-            for album in preflight["albums"]:
-                destination_dir = Path(album["destination"])
-                for asset in album["assets"]:
-                    source = Path(asset["current_path"])
-                    destination = destination_dir / source.name
-                    session.add(
-                        ArchiveOperation(
-                            id=str(uuid.uuid4()),
-                            plan_id=plan_id,
-                            direction=ARCHIVE,
-                            sequence=sequence,
-                            album=album["album"],
-                            asset_id=asset["asset_id"],
-                            source_path=str(source),
-                            destination_path=str(destination),
-                            archive_path=str(destination.relative_to(root)),
-                            expected_sha256=asset["current_sha256"],
-                            byte_size=asset["byte_size"],
-                            # Recorded as a preview only; the real decision is made
-                            # against the devices at apply time.
-                            same_filesystem=album["transfer_method"] == "move",
-                            journal_state=ArchiveState.PLANNED,
-                        )
-                    )
-                    sequence += 1
-            session.commit()
-        return self.get(plan_id)
-
-    def get(self, plan_id: str) -> dict | None:
-        with self._session_factory() as session:
-            plan = session.get(ArchivePlan, plan_id)
-            if plan is None:
-                return None
-            report = _plan_dict(plan)
-        report["operations"] = self.journal.operations(plan_id)
-        return report
-
-    def list(self) -> list[dict]:
-        with self._session_factory() as session:
-            rows = session.scalars(
-                select(ArchivePlan)
-                .where(ArchivePlan.direction == ARCHIVE)
-                .order_by(ArchivePlan.created_at)
-            )
-            return [_plan_dict(row) for row in rows]
-
-    # ── apply ─────────────────────────────────────────────────────────────────
-
-    def apply(
-        self,
-        plan_id: str,
-        *,
-        expected_version: int | None = None,
-        worker_id: str = "archive",
-    ) -> dict:
-        """Archive every item of a plan, then report what happened.
-
-        Items are independent: one failure records its reason and leaves that
-        source in place; the rest of the album continues.
-        """
-        plan = self._require_plan(plan_id)
-        if expected_version is not None and plan["version"] != expected_version:
-            raise ArchiveError(
-                "stale_plan",
-                f"plan {plan_id} is at version {plan['version']}, expected {expected_version}",
-            )
-        if plan["state"] not in APPLYABLE_PLAN_STATES:
-            raise ArchiveError("invalid_state", f"plan {plan_id} is {plan['state']}")
-
-        # One archiver lane: never start while another plan may be half-archived
-        # (concept §16 lock hierarchy).
-        blocking = [row for row in self.journal.incomplete() if row["plan_id"] != plan_id]
-        if blocking:
-            raise ArchiveError(
-                "archive_pending",
-                f"another archive operation is unresolved ({blocking[0]['id']}); recover it first",
-            )
-
-        token = self._claim_plan(plan_id)
-        location = self._location(plan["location_id"])
-        archived = failed = skipped = 0
-        for operation in self.journal.operations(plan_id):
-            if operation["journal_state"] == ArchiveState.COMPLETE:
-                skipped += 1  # repeated apply is a no-op for finished work
-                continue
-            try:
-                if operation["journal_state"] in (ArchiveState.VERIFIED, ArchiveState.REMOVING):
-                    # The bytes are already archived; never transfer them twice.
-                    self._finish(operation, location, token=token, worker_id=worker_id)
-                else:
-                    self._archive_one(operation, location, token=token, worker_id=worker_id)
-                archived += 1
-            except PreconditionFailed as error:
-                self._fail(operation, token, error.code, str(error))
-                failed += 1
-            except Exception as error:  # unexpected: record and stop touching disk
-                self._fail(operation, token, "archive_error", str(error))
-                failed += 1
-        self._prune_empty_sources(plan_id)
-        state = self.journal.sync_plan_state(plan_id)
-        return {
-            "plan_id": plan_id,
-            "archived": archived,
-            "failed": failed,
-            "skipped": skipped,
-            "state": state,
-        }
-
-    def _prune_empty_sources(self, plan_id: str) -> None:
-        """Drop an album folder once every one of its files is archived.
-
-        ``rmdir`` only: a folder that still holds anything at all — an unarchived
-        file, someone else's file, a subfolder — is left exactly as it is.
-        """
-        folders: dict[Path, set[str]] = {}
-        for operation in self.journal.operations(plan_id):
-            folders.setdefault(Path(operation["source_path"]).parent, set()).add(
-                operation["journal_state"]
-            )
-        for folder, states in folders.items():
-            if states == {ArchiveState.COMPLETE}:
-                try:
-                    folder.rmdir()
-                except OSError:
-                    pass  # not empty, or gone already; either way, leave it alone
-
-    def _archive_one(self, operation: dict, location: dict, *, token: int, worker_id: str) -> None:
-        source = Path(operation["source_path"])
-        destination = Path(operation["destination_path"])
-
-        # 1. Intent first — after this point a crash is recoverable from evidence.
-        self.journal.begin(operation["id"], worker_id=worker_id, fencing_token=token)
-        maybe_fault(ArchiveState.TRANSFERRING)
-
-        # 2. Recheck immediately before mutating; the plan's snapshot is not trusted.
-        self._recheck(operation, source, destination, location)
-        destination.parent.mkdir(parents=True, exist_ok=True)
-
-        # 3. Transfer. Same-filesystem is an optimisation, so it has to be proven
-        #    here rather than believed from the plan.
-        same_filesystem = _same_filesystem(source, destination.parent)
-        if same_filesystem:
-            os.rename(source, destination)
-        else:
-            copy_verify_publish(source, destination, operation["expected_sha256"])
-        _fsync_dir(destination.parent)
-
-        # 4. The published file is the archive only once it hashes as recorded.
-        if sha256_file(destination) != operation["expected_sha256"]:
-            raise PreconditionFailed(
-                "archive_mismatch", f"{destination} does not hold the expected bytes"
-            )
-        _append_manifest(destination.parent, _manifest_entry(operation, location))
-        self.journal.transition(
-            operation["id"],
-            ArchiveState.VERIFIED,
-            fencing_token=token,
-            same_filesystem=same_filesystem,
-        )
-        maybe_fault(ArchiveState.VERIFIED)
-
-        # 5. Only now may the active source go.
-        self._finish(self.journal.get(operation["id"]), location, token=token, worker_id=worker_id)
-
-    def _finish(self, operation: dict, location: dict, *, token: int, worker_id: str) -> None:
-        """Drive an item whose archive copy is durable through removal and
-        bookkeeping. Every step is idempotent, so recovery may replay it."""
-        destination = Path(operation["destination_path"])
-        source = Path(operation["source_path"])
-        state = operation["journal_state"]
-
-        if state == ArchiveState.FAILED:
-            # The archive copy is durable even though the attempt ended badly:
-            # re-enter the transfer state so the remaining steps can run.
-            self.journal.transition(operation["id"], ArchiveState.TRANSFERRING, fencing_token=token)
-            state = ArchiveState.TRANSFERRING
-
-        if state == ArchiveState.TRANSFERRING:
-            if sha256_file(destination) != operation["expected_sha256"]:
-                raise PreconditionFailed(
-                    "archive_mismatch", f"{destination} does not hold the expected bytes"
-                )
-            _append_manifest(destination.parent, _manifest_entry(operation, location))
-            self.journal.transition(operation["id"], ArchiveState.VERIFIED, fencing_token=token)
-            state = ArchiveState.VERIFIED
-
-        if state == ArchiveState.VERIFIED:
-            self.journal.transition(operation["id"], ArchiveState.REMOVING, fencing_token=token)
-            maybe_fault(ArchiveState.REMOVING)
-            state = ArchiveState.REMOVING
-
-        if state == ArchiveState.REMOVING:
-            # Re-verify the archived bytes immediately before deleting the original:
-            # this check is the entire justification for the removal.
-            if not destination.exists() or sha256_file(destination) != operation["expected_sha256"]:
-                raise PreconditionFailed(
-                    "archive_unverified", f"{destination} is not a verified archive copy"
-                )
-            self._require_evidence(operation["asset_id"], destination)
-            if source.exists():
-                if source.is_symlink():
-                    raise PreconditionFailed("symlink", f"{source} became a symlink")
-                if sha256_file(source) != operation["expected_sha256"]:
-                    raise PreconditionFailed(
-                        "source_changed", f"{source} changed; it is not ours to remove"
-                    )
-                source.unlink()
-            if source.exists():
-                raise PreconditionFailed("removal_failed", f"{source} is still present")
-            # The dangerous window: active storage no longer holds the file while the
-            # database still points at it.
-            maybe_fault(SOURCE_REMOVED)
-            self._record_archived(operation, location, destination)
-            self.journal.transition(operation["id"], ArchiveState.COMPLETE, fencing_token=token)
-            maybe_fault(ArchiveState.COMPLETE)
-
-    def _recheck(self, operation: dict, source: Path, destination: Path, location: dict) -> None:
-        if not source.exists():
-            raise PreconditionFailed("source_missing", f"source {source} disappeared")
-        if source.is_symlink() or destination.is_symlink():
-            raise PreconditionFailed("symlink", "refusing to archive through a symlink")
-        if destination.exists():
-            raise PreconditionFailed("destination_exists", f"destination {destination} is occupied")
-        root = Path(location["root"])
-        if root not in destination.parents:
-            raise PreconditionFailed(
-                "destination_escape", f"{destination} is outside the archive location {root}"
-            )
-        if not root.is_dir() or not (root / MARKER_NAME).exists():
-            raise PreconditionFailed("location_offline", f"{root} is not the archive medium")
-        if sha256_file(source) != operation["expected_sha256"]:
-            raise PreconditionFailed(
-                "source_changed", f"{source} changed since the plan was approved"
-            )
-        with self._session_factory() as session:
-            asset = session.get(Asset, operation["asset_id"])
-            if asset is None or asset.current_path != str(source):
-                raise PreconditionFailed(
-                    "asset_moved", f"asset {operation['asset_id']} is no longer at {source}"
-                )
-
-    def _require_evidence(self, asset_id: str, source: Path) -> dict:
-        """Review evidence must be durable before the original goes.
-
-        The perceptual hash keeps the asset in the fuzzy index once its bytes are
-        unreachable, and the protected preview is what duplicate review can still
-        look at. Both are read from the freshly verified archive copy, which holds
-        exactly the bytes being archived. A file that cannot be decoded has neither
-        — recorded, not fatal, since its exact hashes remain — but failing to
-        produce a preview from a decodable original stops the removal (concept §9).
-        """
-        DuplicateService(self._session_factory).ensure_phash(asset_id, source=source)
-        preview = ThumbnailService(self._session_factory, self._config).ensure_protected(
-            asset_id, source=source
-        )
-        if preview["state"] == "unavailable":
-            raise PreconditionFailed(
-                "preview_unavailable",
-                f"no durable comparison preview for asset {asset_id} "
-                f"({preview['error_code']})",
-            )
-        return preview
-
-    # ── database ──────────────────────────────────────────────────────────────
-
-    def _record_archived(self, operation: dict, location: dict, destination: Path) -> None:
-        """The original is gone from active storage: drop ``current_path``, close its
-        occurrence, record where the bytes now live, and set availability."""
-        now = _now()
-        with self._session_factory() as session:
-            asset = session.get(Asset, operation["asset_id"])
-            if asset is None:
-                raise PreconditionFailed(
-                    "asset_missing", f"asset {operation['asset_id']} no longer exists"
-                )
-            if asset.current_path:
-                for row in session.scalars(
-                    select(AssetPath).where(
-                        AssetPath.asset_id == asset.id,
-                        AssetPath.path == asset.current_path,
-                        AssetPath.valid_until.is_(None),
-                    )
-                ):
-                    row.valid_until = now
-            recorded = session.scalar(
-                select(AssetPath).where(
-                    AssetPath.asset_id == asset.id, AssetPath.path == str(destination)
-                )
-            )
-            if recorded is None:  # idempotent: recovery may replay this
-                session.add(
-                    AssetPath(
-                        asset_id=asset.id,
-                        path=str(destination),
-                        valid_from=now,
-                        reason="archive",
-                    )
-                )
-            asset.current_path = None
-            asset.availability_state = (
-                "archived_online" if destination.exists() else "archived_offline"
-            )
-            asset.archive_location_id = location["id"]
-            asset.archive_path = operation["archive_path"]
-            asset.state_version += 1
-            asset.updated_at = now
-            session.commit()
-
-    # ── recovery ──────────────────────────────────────────────────────────────
-
-    def recover(self, *, worker_id: str = "archive-recovery") -> dict:
-        """Resolve every incomplete item from journal + disk evidence.
-
-        Idempotent: running it repeatedly converges. Ambiguous (``manual``) work is
-        left exactly as found and keeps blocking unrelated mutations.
-        """
-        results = {"resumed": 0, "completed": 0, "manual": 0}
-        touched: set[str] = set()
-        for verdict in self.journal.classify_all(direction=ARCHIVE):
-            operation = self.journal.get(verdict["operation_id"])
-            touched.add(operation["plan_id"])
-            token = (operation["fencing_token"] or 0) + 1
-            if verdict["classification"] == MANUAL:
-                results["manual"] += 1
-                continue
-            if verdict["classification"] == RESUMABLE:
-                # Nothing was published: discard the debris and let a later apply
-                # retry the item cleanly.
-                _clean_temp_files(Path(operation["destination_path"]).parent)
-                self.journal.transition(operation["id"], ArchiveState.PLANNED, fencing_token=token)
-                results["resumed"] += 1
-                continue
-            location = self._location(self._require_plan(operation["plan_id"])["location_id"])
-            try:
-                self._finish(operation, location, token=token, worker_id=worker_id)
-                results["completed"] += 1
-            except PreconditionFailed as error:
-                self._fail(operation, token, error.code, str(error))
-                results["manual"] += 1
-        for plan_id in touched:
-            self._prune_empty_sources(plan_id)
-            self.journal.sync_plan_state(plan_id)
-        return results
-
-    def recovery_status(self) -> dict:
-        verdicts = self.journal.classify_all(direction=ARCHIVE)
-        return {
-            "operations": verdicts,
-            "manual": [v for v in verdicts if v["classification"] == MANUAL],
-            "blocks_mutation": self.journal.blocks_mutation(),
-        }
-
-    # ── helpers ───────────────────────────────────────────────────────────────
-
-    def _fail(self, operation: dict, token: int, code: str, message: str) -> None:
-        self.journal.transition(
-            operation["id"], ArchiveState.FAILED, fencing_token=token, error=(code, message)
-        )
-
-    def _require_plan(self, plan_id: str) -> dict:
-        with self._session_factory() as session:
-            plan = session.get(ArchivePlan, plan_id)
-            if plan is None:
-                raise ArchiveError("unknown_plan", f"unknown archive plan {plan_id!r}")
-            return _plan_dict(plan)
-
-    def _location(self, location_id: str) -> dict:
-        with self._session_factory() as session:
-            location = session.get(ArchiveLocation, location_id)
-            if location is None:
-                raise ArchiveError("unknown_location", f"unknown archive location {location_id!r}")
-            return {"id": location.id, "root": location.root, "media_id": location.media_id}
-
-    def _claim_plan(self, plan_id: str) -> int:
-        """Bump the plan version and use it as this attempt's fencing token, so a
-        worker from a superseded attempt cannot commit."""
-        with self._session_factory() as session:
-            plan = session.get(ArchivePlan, plan_id)
-            plan.version += 1
-            plan.state = "applying"
-            plan.updated_at = _now()
-            token = plan.version
-            session.commit()
-        return token
-
-
-# ── module helpers ───────────────────────────────────────────────────────────
-
-
-def copy_verify_publish(source: Path, destination: Path, expected_sha256: str) -> None:
-    """Copy to a temporary file beside the destination, prove its bytes, then publish
-    it atomically. The source is never touched, so a failure costs nothing.
-
-    Shared by archiving (library → medium) and restoring (medium → library, US06-04):
-    both need the same promise that a published file is either complete and correct
-    or not there at all.
-    """
-    temp = destination.with_name(f"{TEMP_PREFIX}{uuid.uuid4().hex}{TEMP_SUFFIX}")
-    try:
-        with open(source, "rb") as src, open(temp, "wb") as out:
-            shutil.copyfileobj(src, out, 1024 * 1024)
-            out.flush()
-            os.fsync(out.fileno())
-        if sha256_file(temp) != expected_sha256:
-            raise PreconditionFailed("copy_mismatch", f"{source} copied with wrong bytes")
-        if destination.exists():
-            raise PreconditionFailed(
-                "destination_exists", f"{destination} appeared during the transfer"
-            )
-        # ponytail: rename after an exists() check. The archiver lane is single and
-        # local; use O_EXCL/link-based publish if a second writer ever exists.
-        os.rename(temp, destination)
-    finally:
-        temp.unlink(missing_ok=True)
-
-
-def _same_filesystem(source: Path, destination_dir: Path) -> bool:
-    """Proven at run time from the actual devices, never from the plan's preview."""
-    try:
-        return source.stat().st_dev == destination_dir.stat().st_dev
-    except OSError:
-        return False
-
-
-def _fsync_dir(path: Path) -> None:
-    """Make the directory entry itself durable, so the published name survives a
-    power loss and not just the file's data."""
-    fd = os.open(path, os.O_RDONLY)
-    try:
-        os.fsync(fd)
-    except OSError:
-        pass  # some filesystems refuse directory fsync; the data is already synced
-    finally:
-        os.close(fd)
-
-
-def _manifest_entry(operation: dict, location: dict) -> dict:
-    return {
-        "schema_version": MANIFEST_VERSION,
-        "plan_id": operation["plan_id"],
-        "asset_id": operation["asset_id"],
-        "album": operation["album"],
-        "archive_path": operation["archive_path"],
-        "source_path": operation["source_path"],
-        "sha256": operation["expected_sha256"],
-        "byte_size": operation["byte_size"],
-        "media_id": location["media_id"],
-        "archived_at": _now().isoformat(),
-    }
-
-
-def _append_manifest(directory: Path, entry: dict) -> None:
-    """Append one durable manifest line, skipping an entry that is already there.
-
-    The manifest is written before the source is removed, so it is the medium's own
-    record of what it holds even if the database is lost.
-    """
-    path = directory / MANIFEST_NAME
-    # ponytail: rereads the album manifest per file (O(n²) lines for one album).
-    # Keep an in-memory index per plan if an album ever holds enough files to matter.
-    if path.exists():
-        for line in path.read_text(encoding="utf-8").splitlines():
-            try:
-                existing = json.loads(line)
-            except ValueError:
-                continue
-            if (existing.get("asset_id"), existing.get("sha256")) == (
-                entry["asset_id"],
-                entry["sha256"],
-            ):
-                return
-    with open(path, "a", encoding="utf-8") as handle:
-        handle.write(json.dumps(entry, sort_keys=True) + "\n")
-        handle.flush()
-        os.fsync(handle.fileno())
-    _fsync_dir(directory)
-
-
-def read_manifest(directory: Path) -> list[dict]:
-    """Every manifest entry an archived album directory holds."""
-    path = directory / MANIFEST_NAME
-    if not path.exists():
-        return []
-    entries = []
-    for line in path.read_text(encoding="utf-8").splitlines():
-        try:
-            entries.append(json.loads(line))
-        except ValueError:
-            continue
-    return entries
-
-
-def _clean_temp_files(directory: Path) -> None:
-    """Remove this application's own abandoned transfer temporaries — never any
-    other file (concept §17: startup cleans only recognised stale temporaries)."""
-    if not directory.is_dir():
-        return
-    for temp in directory.glob(f"{TEMP_PREFIX}*{TEMP_SUFFIX}"):
-        temp.unlink(missing_ok=True)
-
-
-def _plan_dict(plan: ArchivePlan) -> dict:
-    return {
-        "id": plan.id,
-        "location_id": plan.location_id,
-        "token": plan.token,
-        "direction": plan.direction,
-        "albums": json.loads(plan.albums) if plan.albums else None,
-        "state": plan.state,
-        "schema_version": plan.schema_version,
-        "asset_count": plan.asset_count,
-        "byte_size": plan.byte_size,
-        "version": plan.version,
-        "worker_id": plan.worker_id,
-        "completed_at": plan.completed_at.isoformat() if plan.completed_at else None,
-        "created_at": plan.created_at.isoformat() if plan.created_at else None,
-    }

photo_pipeline/services/archives.py

@@ -24,15 +24,10 @@ Preflight proves, per concept §9 "Archive preflight":
   probed by writing them, not assumed.
 
 Blocker codes: ``no_library_root``, ``location_offline``, ``wrong_volume``,
-``archive_pending``,
 ``unsafe_destination``, ``destination_not_writable``, ``manifest_unwritable``,
 ``insufficient_capacity``, ``backup_unavailable``, ``lock_conflict``,
 ``rename_pending``, ``empty_scope``, ``destination_collision``,
-``upload_unverified``, ``bytes_changed``, ``file_missing``, ``preview_unavailable``.
+``upload_unverified``, ``bytes_changed``, ``file_missing``.
-
-Preflight also *creates* the durable comparison preview of every asset in scope
-(US06-03): it is the evidence duplicate review falls back on once the original is
-on a medium that may be offline, so it has to exist before the original leaves.
 
 Like upload preflight, the confirmation token is *derived* from the report rather
 than stored: any change to the scope, the bytes, the destination, or the blockers
@@ -62,17 +57,14 @@ from photo_pipeline.jobs.domain_handlers import ARCHIVE_LOCK, LIBRARY_WRITE_LOCK
 from photo_pipeline.models import ArchiveLocation, Asset, UploadBatch, UploadItem
 from photo_pipeline.path_policy import PathPolicyError, is_excluded, normalize_root, resolve_within
 from photo_pipeline.services.albums import album_label
-from photo_pipeline.services.archive_journal import ArchiveJournal
-from photo_pipeline.services.availability import MARKER_NAME, read_marker as _read_marker
-from photo_pipeline.services.availability import refresh as refresh_availability
 from photo_pipeline.services.hashing import sha256_file
 from photo_pipeline.services.jobs import JobService
 from photo_pipeline.services.rename_journal import RenameJournal
-from photo_pipeline.services.thumbnails import ThumbnailService
 from photo_pipeline.services.upload_reports import VERIFIED
 
 PREFLIGHT_VERSION = 1
 TOKEN_PREFIX = f"v{PREFLIGHT_VERSION}"
+MARKER_NAME = ".photo-pipeline-archive.json"
 MANIFEST_NAME = "archive-manifest.json"
 
 # Upload outcomes that prove Immich holds these exact bytes. ``skipped``/``failed``/
@@ -169,10 +161,7 @@ class ArchiveService:
                     location.capabilities = json.dumps(probe["capabilities"])
                 reports.append(self._location_report(location, probe=probe))
             session.commit()
-        # A medium that just appeared or vanished changes what is readable, so the
+            return reports
-        # archived assets are re-derived from the same probe (US06-03).
-        refresh_availability(self._session_factory)
-        return reports
 
     # ── preflight ─────────────────────────────────────────────────────────────
 
@@ -208,12 +197,9 @@ 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 on "
+                    f"{self._config.archive_free_space_reserve_bytes} B reserve do not fit in "
-                    "the medium",
+                    f"{report['capacity']['free_bytes']} B of free space",
                 )
             )
         report["backup"] = self._backup_probe()
@@ -305,13 +291,6 @@ class ArchiveService:
             blockers.append(
                 _issue("rename_pending", "an unresolved rename must be recovered before archiving")
             )
-        if ArchiveJournal(self._session_factory).blocks_mutation():
-            blockers.append(
-                _issue(
-                    "archive_pending",
-                    "an unresolved archive transfer must be recovered before archiving again",
-                )
-            )
         return blockers
 
     def _capacity(self, required: int, probe: dict) -> dict:
@@ -432,10 +411,7 @@ class ArchiveService:
 
     def _album(self, name: str, rows: list[dict], root: Path, *, reachable: bool) -> dict:
         folder = Path(rows[0]["path"]).parent
-        items = sorted(
+        items = sorted((_item(row) for row in rows), key=lambda item: item["current_path"])
-            (self._with_preview(_item(row)) for row in rows),
-            key=lambda item: item["current_path"],
-        )
         blocked = [item for item in items if item["blockers"]]
         blockers: list[dict] = []
 
@@ -471,30 +447,6 @@ class ArchiveService:
             "assets": items,
         }
 
-    def _with_preview(self, item: dict) -> dict:
-        """Create the durable comparison preview while the original is still here.
-
-        This is the last moment it can be made: once the file is archived and the
-        medium leaves, only the retained preview can answer "is this new photo the
-        same picture?". An original that cannot be decoded at all has no preview to
-        keep — its hashes and metadata stay the evidence — but a preview that fails
-        for any other reason blocks the archive (concept §9).
-        """
-        preview = self._previews().ensure_protected(item["asset_id"])
-        item["preview"] = preview
-        if preview["state"] == "unavailable" and not item["blockers"]:
-            item["blockers"].append(
-                _issue(
-                    "preview_unavailable",
-                    f"a durable comparison preview of {item['current_path']} could not be "
-                    f"created ({preview['error_code']})",
-                )
-            )
-        return item
-
-    def _previews(self) -> ThumbnailService:
-        return ThumbnailService(self._session_factory, self._config)
-
 
 # ── internals ────────────────────────────────────────────────────────────────
 
@@ -551,6 +503,13 @@ def _transfer_method(folder: Path, root: Path) -> str:
     return "copy_verify_remove"
 
 
+def _read_marker(root: Path) -> dict | None:
+    try:
+        return json.loads((root / MARKER_NAME).read_text(encoding="utf-8"))
+    except (OSError, ValueError):
+        return None
+
+
 def _probe_write(path: Path, payload: bytes, *, keep: bool = False) -> str | None:
     """Write ``payload`` to ``path``; return the failure detail or ``None``."""
     try:

photo_pipeline/services/availability.py

@@ -1,124 +0,0 @@
-"""Where an asset's bytes are right now (US06-03).
-
-Archiving removes the original from the active library but never removes the
-asset: its identity, hashes, decisions, and evidence stay. This module is the one
-place that answers "can these bytes be read, and if not, why" so inventory,
-duplicate review, thumbnails, and the archive service all give the same answer.
-
-States (concept §9):
-
-- ``active``             — the original is in the active library;
-- ``archived_online``    — the recorded medium is mounted and holds the file;
-- ``archived_offline``   — archived, but the medium is not available right now;
-- ``missing_unexpected`` — neither an active path nor the recorded archive
-  location explains the absence. This is the state that must never be confused
-  with ``archived_offline``: an unmounted disk is normal, a mounted disk with a
-  hole in it is not.
-
-A medium is identified by its marker file, never by its mountpoint, so a
-different disk mounted at the recorded root is offline rather than accepted.
-"""
-
-from __future__ import annotations
-
-import json
-from collections import Counter
-from datetime import datetime, timezone
-from pathlib import Path
-
-from sqlalchemy import select
-from sqlalchemy.orm import Session, sessionmaker
-
-from photo_pipeline.models import ArchiveLocation, Asset
-
-ACTIVE = "active"
-ARCHIVED_ONLINE = "archived_online"
-ARCHIVED_OFFLINE = "archived_offline"
-MISSING_UNEXPECTED = "missing_unexpected"
-ARCHIVED = (ARCHIVED_ONLINE, ARCHIVED_OFFLINE)
-
-MARKER_NAME = ".photo-pipeline-archive.json"
-
-
-def read_marker(root: Path) -> dict | None:
-    """The medium's identity marker, or ``None`` when it is not readable."""
-    try:
-        return json.loads((root / MARKER_NAME).read_text(encoding="utf-8"))
-    except (OSError, ValueError):
-        return None
-
-
-def location_online(location: ArchiveLocation) -> bool:
-    """True only when the *recorded* medium is mounted at its root."""
-    marker = read_marker(Path(location.root))
-    return bool(marker) and marker.get("media_id") == location.media_id
-
-
-def archive_file(session: Session, asset: Asset) -> Path | None:
-    """The archived file's absolute path, whether or not the medium is mounted."""
-    if not asset.archive_location_id or not asset.archive_path:
-        return None
-    location = session.get(ArchiveLocation, asset.archive_location_id)
-    if location is None:
-        return None
-    return Path(location.root) / asset.archive_path
-
-
-def readable_path(session: Session, asset: Asset) -> Path | None:
-    """A path whose bytes can be read now: the active file, else the archive copy."""
-    if asset.current_path and Path(asset.current_path).exists():
-        return Path(asset.current_path)
-    archived = archive_file(session, asset)
-    if archived is None:
-        return None
-    location = session.get(ArchiveLocation, asset.archive_location_id)
-    if not location_online(location) or not archived.exists():
-        return None
-    return archived
-
-
-def state_of(session: Session, asset: Asset, *, online: dict[str, bool] | None = None) -> str:
-    """The availability this asset's storage actually justifies right now."""
-    if asset.current_path:
-        return ACTIVE if Path(asset.current_path).exists() else MISSING_UNEXPECTED
-    if not asset.archive_location_id:
-        return MISSING_UNEXPECTED if asset.availability_state != ACTIVE else ACTIVE
-    location = session.get(ArchiveLocation, asset.archive_location_id)
-    if location is None:
-        return MISSING_UNEXPECTED
-    reachable = (
-        online[location.id] if online and location.id in online else location_online(location)
-    )
-    if not reachable:
-        return ARCHIVED_OFFLINE
-    archived = archive_file(session, asset)
-    # The medium is mounted and identified: the file is either there, or it is
-    # genuinely gone — that is not "offline", it needs a human.
-    return ARCHIVED_ONLINE if archived and archived.exists() else MISSING_UNEXPECTED
-
-
-def refresh(session_factory: sessionmaker) -> dict[str, int]:
-    """Re-derive availability for every archived asset from the media themselves.
-
-    Only archived assets are probed: whether an *active* file is present is the
-    inventory scan's job and costs one stat per library file. Each medium is
-    probed once, not once per asset.
-    """
-    counts: Counter[str] = Counter()
-    now = datetime.now(timezone.utc)
-    with session_factory() as session:
-        online = {
-            location.id: location_online(location)
-            for location in session.scalars(select(ArchiveLocation))
-        }
-        for asset in session.scalars(
-            select(Asset).where(Asset.archive_location_id.is_not(None))
-        ):
-            state = state_of(session, asset, online=online)
-            counts[state] += 1
-            if state != asset.availability_state:
-                asset.availability_state = state
-                asset.state_version += 1
-                asset.updated_at = now
-        session.commit()
-    return dict(counts)

photo_pipeline/services/duplicates.py

@@ -11,12 +11,6 @@ Detection runs in two categories:
   band (NEAR/SIMILAR). These are review candidates: never decided automatically, and
   negative-linked pairs are suppressed so a rejected pair is not re-suggested.
 
-Archived assets stay in both indexes (US06-03): a new active copy of an archived
-original is recognised through its hashes even while the medium is offline, and
-cluster review falls back to the retained protected preview plus hash evidence.
-An exact/pixel match links straight to the archived canonical; a perceptual match
-is a review candidate that names the medium to mount for a pixel-level decision.
-
 Decisions (``canonical`` / ``not_duplicate`` / ``deferred``) persist with evidence,
 use optimistic version checks, are reversible, and can never form a canonical cycle.
 A new content-identical member of an already-decided cluster inherits the established
@@ -40,14 +34,12 @@ from sqlalchemy import func, select
 from sqlalchemy.orm import sessionmaker
 
 from photo_pipeline.models import (
-    ArchiveLocation,
     Asset,
     DuplicateCluster,
     DuplicateMember,
     DuplicateNegativeLink,
-    Thumbnail,
 )
-from photo_pipeline.services import availability, hashing
+from photo_pipeline.services import hashing
 
 NEAR_MAX = 5
 SIMILAR_MAX = 10
@@ -135,18 +127,18 @@ class DuplicateService:
 
     # ── perceptual hash backfill ───────────────────────────────────────────
     def ensure_phashes(self) -> int:
-        """Hash whatever is readable now — an archived asset keeps the hash it
-        already has, and gains one whenever its medium happens to be mounted."""
         updated = 0
         with self._session_factory() as session:
-            assets = session.execute(select(Asset)).scalars()
+            assets = session.execute(
+                select(Asset).where(
+                    Asset.availability_state == "active",
+                    Asset.current_path.isnot(None),
+                )
+            ).scalars()
             for asset in assets:
                 if asset.phash is not None and asset.phash_version == hashing.PHASH_VERSION:
                     continue
-                source = availability.readable_path(session, asset)
+                value = hashing.safe_phash(asset.current_path)
-                if source is None:
-                    continue
-                value = hashing.safe_phash(str(source))
                 if value is not None:
                     asset.phash = value
                     asset.phash_version = hashing.PHASH_VERSION
@@ -154,40 +146,20 @@ class DuplicateService:
             session.commit()
         return updated
 
-    def ensure_phash(self, asset_id: str, *, source=None) -> str | None:
-        """Backfill one asset's perceptual hash while its bytes are still readable.
-
-        Archiving calls this before the original leaves — passing the archive copy
-        as ``source``, since the database does not point at it yet — because an
-        asset without a pHash silently drops out of the fuzzy index the moment its
-        medium is away.
-        """
-        with self._session_factory() as session:
-            asset = session.get(Asset, asset_id)
-            if asset is None:
-                return None
-            if asset.phash is not None and asset.phash_version == hashing.PHASH_VERSION:
-                return asset.phash
-            source = source or availability.readable_path(session, asset)
-            if source is None:
-                return None
-            value = hashing.safe_phash(str(source))
-            if value is not None:
-                asset.phash = value
-                asset.phash_version = hashing.PHASH_VERSION
-                session.commit()
-            return value
-
     # ── detection ──────────────────────────────────────────────────────────
     def detect(self) -> DetectionReport:
         self.ensure_phashes()
         now = datetime.now(timezone.utc)
         report = DetectionReport()
         with self._session_factory() as session:
-            # Every known asset stays in the indexes, archived or not: a copy of an
+            assets = list(
-            # archived original must be recognised as a duplicate rather than
+                session.execute(
-            # treated as a new photo (concept §9, invariant 12).
+                    select(Asset).where(
-            assets = list(session.execute(select(Asset)).scalars())
+                        Asset.availability_state == "active",
+                        Asset.current_path.isnot(None),
+                    )
+                ).scalars()
+            )
             by_id = {a.id: a for a in assets}
             negatives = {
                 _pair(link.asset_a, link.asset_b)
@@ -459,19 +431,10 @@ class DuplicateService:
 
     @staticmethod
     def _recommend_canonical(ids, by_id) -> str:
-        # ponytail: largest file, then the archived copy, then path as a
+        # ponytail: largest file, path as deterministic tie-break. The concept's
-        # deterministic tie-break. Archived wins ties because it is the reviewed,
+        # richer policy (resolution, least recompression, metadata richness) lands
-        # uploaded original — a fresh active copy must not demote it to a variant.
+        # with the review UI story.
-        # The concept's richer policy (resolution, least recompression, metadata
+        return max(ids, key=lambda i: (by_id[i].byte_size or 0, by_id[i].current_path or ""))
-        # richness) lands with the review UI story.
-        return max(
-            ids,
-            key=lambda i: (
-                by_id[i].byte_size or 0,
-                by_id[i].availability_state in availability.ARCHIVED,
-                by_id[i].current_path or by_id[i].archive_path or "",
-            ),
-        )
 
     def _apply_canonical(self, session, cluster, ids, canonical_id):
         for member in session.execute(
@@ -544,15 +507,9 @@ class DuplicateService:
                         "current_path": asset.current_path if asset else None,
                         "byte_size": asset.byte_size if asset else None,
                         "phash": asset.phash if asset else None,
-                        **self._offline_evidence(session, asset),
                     }
                 )
             members.sort(key=lambda m: m["asset_id"])
-            # A full-resolution comparison of an offline original is impossible; the
-            # UI asks for that named medium instead of guessing (concept §9).
-            mount_required = sorted(
-                {m["archive_location"] for m in members if m["requires_mount"]}
-            )
             return {
                 "id": cluster.id,
                 "method": cluster.method,
@@ -562,54 +519,9 @@ class DuplicateService:
                 "canonical_asset_id": cluster.canonical_asset_id,
                 "version": cluster.version,
                 "requires_confirmation": cluster.method == Method.PERCEPTUAL.value,
-                "mount_required": mount_required,
                 "members": members,
             }
 
-    def _offline_evidence(self, session, asset: Asset | None) -> dict:
-        """What review can still rely on when a member's original is not readable."""
-        if asset is None:
-            return {
-                "availability_state": None,
-                "archive_location": None,
-                "archive_location_id": None,
-                "archive_path": None,
-                "preview": {"state": "missing", "protected": False},
-                "requires_mount": False,
-            }
-        location = (
-            session.get(ArchiveLocation, asset.archive_location_id)
-            if asset.archive_location_id
-            else None
-        )
-        preview = self._preview_evidence(session, asset.id)
-        archived = asset.availability_state in availability.ARCHIVED
-        return {
-            "availability_state": asset.availability_state,
-            "archive_location": location.name if location else None,
-            "archive_location_id": asset.archive_location_id,
-            "archive_path": asset.archive_path,
-            "preview": preview,
-            # Offline archived members can still be compared through their retained
-            # preview and hash evidence; only pixel-level review needs the medium.
-            "requires_mount": archived
-            and asset.availability_state == availability.ARCHIVED_OFFLINE
-            and bool(location),
-        }
-
-    @staticmethod
-    def _preview_evidence(session, asset_id: str) -> dict:
-        rows = list(
-            session.execute(select(Thumbnail).where(Thumbnail.asset_id == asset_id)).scalars()
-        )
-        ready = [r for r in rows if r.state == "ready" and r.path]
-        if ready:
-            best = max(ready, key=lambda r: (bool(r.protected), r.size or 0))
-            return {"state": "ready", "protected": bool(best.protected), "size": best.size}
-        if rows:
-            return {"state": "unsupported", "protected": False, "size": rows[0].size}
-        return {"state": "missing", "protected": False, "size": None}
-
     # ── decisions ────────────────────────────────────────────────────────────
     def decide(
         self,

photo_pipeline/services/inventory.py

@@ -13,11 +13,8 @@ renames. Every discovered or absent path is classified as one occurrence:
 - ``missing``          — a known active asset whose file is gone (kept, flagged).
 
 Missing files are never pruned (that would break identity); the asset is retained
-with ``missing_at`` set and its availability becomes ``missing_unexpected`` —
+with ``missing_at`` set. Archived assets are left untouched. Rescanning unchanged
-nothing explains where the bytes went. Archived assets are left untouched: their
+input makes no durable change.
-absence from the active roots is expected, and each scan re-derives whether their
-medium is reachable (:mod:`photo_pipeline.services.availability`). Rescanning
-unchanged input makes no durable change.
 
 Extracted from photo_analyzer.discover_photos/reconcile_moved/prune_missing
 (see donor_ledger.yaml: pa-discovery, pa-prune-missing).
@@ -33,12 +30,12 @@ from enum import Enum
 from pathlib import Path
 from typing import Iterable
 
-from sqlalchemy import func, or_, select
+from sqlalchemy import func, select
 from sqlalchemy.orm import Session, sessionmaker
 
 from photo_pipeline import path_policy
 from photo_pipeline.models import Asset, AssetPath
-from photo_pipeline.services import availability, hashing
+from photo_pipeline.services import hashing
 
 
 class Occurrence(str, Enum):
@@ -63,8 +60,6 @@ def _asset_dict(asset: Asset) -> dict:
         "id": asset.id,
         "current_path": asset.current_path,
         "availability_state": asset.availability_state,
-        "archive_location_id": asset.archive_location_id,
-        "archive_path": asset.archive_path,
         "byte_size": asset.byte_size,
         "current_sha256": asset.current_sha256,
         "pixel_sha256": asset.pixel_sha256,
@@ -106,9 +101,7 @@ class InventoryService:
                 result.asset_ids[str(path)] = asset.id
 
             for asset in assets:
-                # Archived assets are explained by their location, not by the active
+                if asset.availability_state != "active" or asset.id in seen_ids:
-                # roots: a scan must never prune or flag them (concept §9).
-                if asset.availability_state in availability.ARCHIVED or asset.id in seen_ids:
                     continue
                 if asset.current_path and asset.current_path not in discovered_paths:
                     if not Path(asset.current_path).exists():
@@ -116,18 +109,10 @@ class InventoryService:
                             asset.missing_at = now
                             asset.state_version += 1
                             asset.updated_at = now
-                        # Nothing explains this absence — it is not an offline medium.
-                        if asset.availability_state != availability.MISSING_UNEXPECTED:
-                            asset.availability_state = availability.MISSING_UNEXPECTED
-                            asset.state_version += 1
-                            asset.updated_at = now
                         result.occurrences[asset.current_path] = Occurrence.MISSING.value
 
             session.commit()
 
-        # Media may have been mounted or removed since the last scan.
-        availability.refresh(self._session_factory)
-
         result.counts = dict(Counter(result.occurrences.values()))
         return result
 
@@ -147,11 +132,7 @@ class InventoryService:
             if availability:
                 stmt = stmt.where(Asset.availability_state == availability)
             if query:
-                like = f"%{query}%"
+                stmt = stmt.where(Asset.current_path.like(f"%{query}%"))
-                # An archived asset has no active path; it is searched where it lives.
-                stmt = stmt.where(
-                    or_(Asset.current_path.like(like), Asset.archive_path.like(like))
-                )
             total = session.scalar(select(func.count()).select_from(stmt.subquery()))
             rows = session.execute(
                 stmt.order_by(Asset.current_path).limit(limit).offset(offset)
@@ -188,7 +169,6 @@ class InventoryService:
                 self._open_path(session, existing.id, path_str, now, occ.value)
             if existing.missing_at is not None:
                 existing.missing_at = None
-                existing.availability_state = availability.ACTIVE
                 existing.state_version += 1
                 existing.updated_at = now
             return existing, occ
@@ -209,7 +189,6 @@ class InventoryService:
             moved_from.current_path = path_str
             moved_from.byte_size = size
             moved_from.missing_at = None
-            moved_from.availability_state = availability.ACTIVE
             moved_from.state_version += 1
             moved_from.updated_at = now
             self._open_path(session, moved_from.id, path_str, now, Occurrence.MOVED.value)

photo_pipeline/services/rename_apply.py

@@ -83,13 +83,12 @@ def _now() -> datetime:
     return datetime.now(timezone.utc)
 
 
-def maybe_fault(state: str) -> None:
+def _maybe_fault(state: str) -> None:
     """Test-only crash barrier (concept §18 fault injection).
 
     When ``PHOTO_PIPELINE_FAULT_AFTER`` names a journal state, the process dies
     abruptly the moment that state has been persisted — modelling a real kill at
-    exactly that transition. Never set outside tests. Shared with the archive
+    exactly that transition. Never set outside tests.
-    transfer journal (US06-02), which uses the same env var and its own state names.
     """
     if os.environ.get("PHOTO_PIPELINE_FAULT_AFTER") == state:
         os._exit(9)
@@ -173,7 +172,7 @@ class RenameApplyService:
 
         # 1. Intent first — after this point a crash is recoverable from evidence.
         self.journal.begin(operation["id"], worker_id=worker_id, fencing_token=token)
-        maybe_fault(JournalState.MOVING)
+        _maybe_fault(JournalState.MOVING)
 
         # 2. Recheck preconditions immediately before mutating, never trusting the
         #    plan's snapshot: files can change between preview and confirmation.
@@ -188,21 +187,21 @@ class RenameApplyService:
             os.rename(source, destination)
 
         self.journal.transition(operation["id"], JournalState.MOVED, fencing_token=token)
-        maybe_fault(JournalState.MOVED)
+        _maybe_fault(JournalState.MOVED)
 
         # 4. Database: stable IDs keep their identity, paths are re-pointed and the
         #    old occurrence is closed — all in one transaction.
         self._reconcile_paths(operation, source, destination)
         self.journal.transition(operation["id"], JournalState.DATABASE_UPDATED, fencing_token=token)
-        maybe_fault(JournalState.DATABASE_UPDATED)
+        _maybe_fault(JournalState.DATABASE_UPDATED)
 
         # 5. Postconditions: the bytes really are at the new paths.
         self._verify(operation, destination)
         self.journal.transition(operation["id"], JournalState.VERIFIED, fencing_token=token)
-        maybe_fault(JournalState.VERIFIED)
+        _maybe_fault(JournalState.VERIFIED)
 
         self.journal.transition(operation["id"], JournalState.COMPLETE, fencing_token=token)
-        maybe_fault(JournalState.COMPLETE)
+        _maybe_fault(JournalState.COMPLETE)
 
     def _recheck(self, operation: dict, source: Path, destination: Path) -> None:
         if not source.exists():

photo_pipeline/services/restores.py

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

photo_pipeline/services/thumbnails.py

@@ -8,11 +8,6 @@ an EXIF-only edit reuses the file while a real pixel change invalidates it; writ
 are atomic and the cache is bounded by an LRU quota. Failures are persisted as typed
 errors so a broken original is not retried on every request.
 
-An archived asset is served from its medium when that medium is mounted, and from
-its *protected* preview when it is not (US06-03). Protected previews are evidence,
-not cache: the quota never evicts them, because the original they describe may be
-unreachable when duplicate review needs it.
-
 Reuses photo_analyzer.prepare_image decode/resize/HEIC handling, adding the missing
 EXIF-orientation step, WebP output, and a managed cache (donor_ledger.yaml:
 pa-imaging).
@@ -31,7 +26,6 @@ from sqlalchemy.orm import sessionmaker
 from photo_pipeline import path_policy
 from photo_pipeline.config import Config
 from photo_pipeline.models import Asset, Thumbnail
-from photo_pipeline.services import availability
 
 # Best-effort HEIC support: registered only if the optional decoder is installed.
 try:  # pragma: no cover - depends on an optional native dependency
@@ -44,8 +38,6 @@ except Exception:  # pragma: no cover
 SIZES = (256, 512, 1280)
 THUMB_VERSION = 1
 THUMB_FORMAT = "webp"
-# The size kept as durable comparison evidence for archived assets (concept §9).
-PROTECTED_SIZE = 1280
 
 
 class ThumbnailError(RuntimeError):
@@ -94,12 +86,7 @@ class ThumbnailService:
         self._config = config
         self._cache_dir = config.thumbnail_cache_dir
 
-    def generate(
+    def generate(self, asset_id: str, size: int) -> Path:
-        self, asset_id: str, size: int, *, protected: bool = False, source: Path | None = None
-    ) -> Path:
-        """Render (or reuse) a preview. ``source`` overrides where the bytes are read
-        from — the archiver passes its verified archive copy, which the database does
-        not yet point at while the transfer is still in flight."""
         if size not in SIZES:
             raise InvalidSize(f"size must be one of {SIZES}")
 
@@ -107,7 +94,9 @@ class ThumbnailService:
             asset = session.get(Asset, asset_id)
             if asset is None:
                 raise ThumbnailNotFound(f"unknown asset {asset_id}")
-            archived = asset.availability_state in availability.ARCHIVED
+            if asset.availability_state != "active" or not asset.current_path:
+                raise ThumbnailUnavailable(f"asset {asset_id} has no active file")
+            self._validate_path(asset.current_path)
             cache_key = self._cache_key(asset, size)
 
             row = session.get(Thumbnail, cache_key)
@@ -118,18 +107,9 @@ class ThumbnailService:
                     )
                 if row.path and Path(row.path).exists():
                     _touch(row.path)
-                    if protected and not row.protected:
-                        self._protect(cache_key)
                     return Path(row.path)
 
-            # An archived original is read from its medium; when that medium is not
+            source = asset.current_path
-            # mounted the retained preview above is the only evidence there is.
-            source = source or availability.readable_path(session, asset)
-            if source is None:
-                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
 
         # Rendering happens outside the DB session (no transaction held during I/O).
         try:
@@ -139,61 +119,10 @@ class ThumbnailService:
                 self._record_error(cache_key, asset_id, size, error.code)
             raise
 
-        # Archived assets keep their preview permanently: it is the comparison
+        self._record_ready(cache_key, asset_id, size, rendered)
-        # evidence that survives the original leaving active storage.
-        self._record_ready(cache_key, asset_id, size, rendered, protected=protected or archived)
         self._enforce_quota(keep=rendered["path"])
         return Path(rendered["path"])
 
-    def ensure_protected(self, asset_id: str, *, source: Path | None = None) -> dict:
-        """Produce (or confirm) the durable comparison preview for an asset.
-
-        Returns evidence rather than raising, because the caller — archive
-        preflight and the transfer itself — decides what an unrenderable original
-        means. ``unsupported`` is a recorded property of the file, not a failure of
-        the policy: its hashes and metadata remain the comparison evidence.
-        """
-        try:
-            path = self.generate(asset_id, PROTECTED_SIZE, protected=True, source=source)
-        except tuple(_PERSISTED_ERRORS) as error:
-            return {"state": "unsupported", "error_code": error.code, "path": None}
-        except ThumbnailError as error:
-            return {"state": "unavailable", "error_code": error.code, "path": None}
-        return {"state": "ready", "error_code": None, "path": str(path)}
-
-    def evidence(self, asset_id: str) -> dict:
-        """What durable preview this asset has right now, without rendering."""
-        with self._session_factory() as session:
-            rows = list(
-                session.execute(
-                    select(Thumbnail).where(Thumbnail.asset_id == asset_id)
-                ).scalars()
-            )
-        for row in rows:
-            if row.state == "ready" and row.path and Path(row.path).exists():
-                return {
-                    "state": "ready",
-                    "protected": bool(row.protected),
-                    "size": row.size,
-                    "error_code": None,
-                }
-        for row in rows:
-            if row.state == "error":
-                return {
-                    "state": "unsupported",
-                    "protected": False,
-                    "size": row.size,
-                    "error_code": row.error_code,
-                }
-        return {"state": "missing", "protected": False, "size": None, "error_code": None}
-
-    def _protect(self, cache_key: str) -> None:
-        with self._session_factory() as session:
-            row = session.get(Thumbnail, cache_key)
-            if row is not None:
-                row.protected = True
-                session.commit()
-
     # ── path safety ──────────────────────────────────────────────────────────
     def _validate_path(self, current_path: str) -> None:
         path = Path(current_path)
@@ -255,9 +184,7 @@ class ThumbnailService:
         }
 
     # ── persistence ────────────────────────────────────────────────────────────
-    def _record_ready(
+    def _record_ready(self, cache_key: str, asset_id: str, size: int, rendered: dict) -> None:
-        self, cache_key: str, asset_id: str, size: int, rendered: dict, *, protected: bool = False
-    ) -> None:
         with self._session_factory() as session:
             session.merge(
                 Thumbnail(
@@ -270,7 +197,6 @@ class ThumbnailService:
                     width=rendered["width"],
                     height=rendered["height"],
                     format=rendered["format"],
-                    protected=protected,
                 )
             )
             try:
@@ -304,10 +230,6 @@ class ThumbnailService:
         if total <= quota:
             return
         files.sort(key=lambda f: f.stat().st_mtime)  # least-recently-used first
-        # Protected previews are evidence, not cache: an archived original cannot be
-        # re-rendered once its medium is away, so eviction never touches them.
-        protected = self._protected_paths()
-        files = [f for f in files if str(f) not in protected]
         keep_path = str(Path(keep)) if keep else None
         evicted: list[str] = []
         for f in files:
@@ -325,16 +247,6 @@ class ThumbnailService:
         if evicted:
             self._forget(evicted)
 
-    def _protected_paths(self) -> set[str]:
-        with self._session_factory() as session:
-            return {
-                row.path
-                for row in session.execute(
-                    select(Thumbnail).where(Thumbnail.protected.is_(True))
-                ).scalars()
-                if row.path
-            }
-
     def _forget(self, paths: list[str]) -> None:
         with self._session_factory() as session:
             rows = session.execute(

tests/e2e/_pipeline_harness.py

@@ -393,125 +393,6 @@ 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

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

tests/e2e/test_phase_f_pipeline.py

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

tests/e2e/test_traceability.py

@@ -14,7 +14,6 @@ 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():
@@ -33,13 +32,6 @@ def test_all_phase_e_stories_are_mapped():
     assert PHASE_E_STORIES <= set(MAP)
 
 
-def test_all_phase_f_stories_are_mapped():
-    """US06-06 acceptance: every archive-lifecycle story, destination through
-    restore, is tied to automated tests — archiving is the only stage that removes
-    an original."""
-    assert PHASE_F_STORIES <= set(MAP)
-
-
 def test_every_mapped_test_file_exists_and_is_nonempty():
     for story, files in MAP.items():
         assert files, f"{story} maps to no tests"

tests/integration/test_archive_recovery.py

@@ -1,313 +0,0 @@
-"""Recovering interrupted archive transfers (US06-02).
-
-The crash tests are real: a child process applies an archive plan and is killed by
-the ``PHOTO_PIPELINE_FAULT_AFTER`` barrier at each persisted transition in turn —
-including the moment immediately after an active source has been unlinked. The
-parent then reopens the database and asserts the one invariant archiving exists to
-uphold: **no verified file is ever lost, and no source is removed without a durable,
-byte-identical archive copy.**
-
-Both transfer paths are exercised: the same-filesystem atomic move and (with the
-child forcing the device comparison) the cross-filesystem copy/verify/publish.
-"""
-
-import os
-import subprocess
-import sys
-import uuid
-from datetime import datetime, timezone
-from pathlib import Path
-
-import pytest
-from sqlalchemy import select
-
-from photo_pipeline.config import Config
-from photo_pipeline.db import create_db_engine, create_session_factory, run_migrations
-from photo_pipeline.models import Asset, UploadBatch, UploadItem
-from photo_pipeline.services.archive_journal import FORWARD, MANUAL, RESUMABLE, ArchiveState
-from photo_pipeline.services.archive_transfer import (
-    MANIFEST_NAME,
-    ArchiveTransferService,
-    read_manifest,
-)
-from photo_pipeline.services.archives import ArchiveService
-from photo_pipeline.services.hashing import sha256_file
-
-pytestmark = pytest.mark.phase_f
-
-REPO = Path(__file__).resolve().parents[2]
-NOW = datetime(2026, 1, 1, tzinfo=timezone.utc)
-
-# Barriers, in the order the transfer persists them. ``source_removed`` is the one
-# that matters most: the original is already gone at that point.
-BARRIERS = [
-    ArchiveState.TRANSFERRING,
-    ArchiveState.VERIFIED,
-    ArchiveState.REMOVING,
-    "source_removed",
-    ArchiveState.COMPLETE,
-]
-
-# A child that applies the plan and dies at the configured barrier. ``force_copy``
-# makes it take the cross-filesystem path without a second real volume.
-APPLY_SCRIPT = """
-import sys
-sys.path.insert(0, {repo!r})
-from photo_pipeline.config import Config
-from photo_pipeline.db import create_db_engine, create_session_factory
-from photo_pipeline.services import archive_transfer
-
-db_url, data_dir, lib, plan_id, force_copy = sys.argv[1:6]
-if force_copy == "1":
-    archive_transfer._same_filesystem = lambda *args: False
-config = Config.from_env(
-    {{"PHOTO_PIPELINE_DATA_DIR": data_dir, "PHOTO_PIPELINE_LIBRARY_ROOTS": lib}}
-)
-sf = create_session_factory(create_db_engine(db_url))
-archive_transfer.ArchiveTransferService(sf, config=config).apply(plan_id)
-"""
-
-
-# ── environment ──────────────────────────────────────────────────────────────
-
-
-def _env(tmp_path):
-    (tmp_path / "data").mkdir(exist_ok=True)
-    lib = tmp_path / "lib"
-    lib.mkdir(exist_ok=True)
-    archive = tmp_path / "archive"
-    archive.mkdir(exist_ok=True)
-    config = Config.from_env(
-        {
-            "PHOTO_PIPELINE_DATA_DIR": str(tmp_path / "data"),
-            "PHOTO_PIPELINE_LIBRARY_ROOTS": str(lib),
-            "PHOTO_PIPELINE_ARCHIVE_FREE_SPACE_RESERVE_BYTES": "0",
-        }
-    )
-    run_migrations(config.database_url)
-    return config, create_session_factory(create_db_engine(config.database_url)), lib, archive
-
-
-def _album(sf, lib, album="rome", names=("a.jpg", "b.jpg")):
-    folder = lib / album
-    folder.mkdir(parents=True, exist_ok=True)
-    with sf() as session:
-        batch_id = str(uuid.uuid4())
-        session.add(
-            UploadBatch(
-                id=batch_id,
-                album=album,
-                folder=str(folder),
-                album_name=album,
-                state="succeeded",
-                preflight_token="v1:test",
-                outcome_state="verified",
-                created_at=NOW,
-            )
-        )
-        for name in names:
-            path = folder / name
-            path.write_bytes(f"{album}/{name} content".encode() * 8)
-            asset_id = str(uuid.uuid4())
-            session.add(
-                Asset(
-                    id=asset_id,
-                    original_path=str(path),
-                    current_path=str(path),
-                    discovered_at=NOW,
-                    hash_version=1,
-                    byte_size=path.stat().st_size,
-                    current_sha256=sha256_file(path),
-                )
-            )
-            session.add(
-                UploadItem(
-                    batch_id=batch_id,
-                    asset_id=asset_id,
-                    path=str(path),
-                    sha256=sha256_file(path),
-                    sha1="0" * 40,
-                    state="sent",
-                    outcome="uploaded",
-                )
-            )
-        session.commit()
-    return folder
-
-
-def _plan(sf, config, archive, albums=None):
-    location = ArchiveService(sf, config=config).register("external", str(archive))
-    token = ArchiveService(sf, config=config).preflight(location["id"], albums)["token"]
-    service = ArchiveTransferService(sf, config=config)
-    return service, service.create(location["id"], albums, token=token)
-
-
-def _crash_during_apply(config, tmp_path, lib, plan, barrier, *, force_copy=False):
-    script = tmp_path / f"apply_{barrier}.py"
-    script.write_text(APPLY_SCRIPT.format(repo=str(REPO)))
-    env = dict(os.environ)
-    env["PHOTO_PIPELINE_FAULT_AFTER"] = barrier
-    result = subprocess.run(
-        [
-            sys.executable,
-            str(script),
-            config.database_url,
-            str(tmp_path / "data"),
-            str(lib),
-            plan["id"],
-            "1" if force_copy else "0",
-        ],
-        env=env,
-        capture_output=True,
-    )
-    assert result.returncode in (9, -9), (
-        f"child should have been killed at {barrier}, got {result.returncode}: "
-        f"{result.stderr.decode(errors='replace')[-400:]}"
-    )
-
-
-def _contents(*roots):
-    return sorted(
-        path.read_bytes()
-        for root in roots
-        for path in root.rglob("*")
-        if path.is_file() and path.name != MANIFEST_NAME and not path.name.startswith(".")
-    )
-
-
-def _reopen(config):
-    return create_session_factory(create_db_engine(config.database_url))
-
-
-# ── fault injection at every persisted transition ────────────────────────────
-
-
-@pytest.mark.parametrize("barrier", BARRIERS)
-@pytest.mark.parametrize("force_copy", [False, True], ids=["move", "copy"])
-def test_a_crash_at_every_transition_loses_nothing_and_recovers(tmp_path, barrier, force_copy):
-    config, sf, lib, archive = _env(tmp_path)
-    _album(sf, lib)
-    service, plan = _plan(sf, config, archive)
-    before = _contents(lib)
-
-    _crash_during_apply(config, tmp_path, lib, plan, barrier, force_copy=force_copy)
-
-    # Every file still exists somewhere: the crash may not have cost a single byte.
-    assert set(before) <= set(_contents(lib, archive)), f"content lost at {barrier}"
-
-    reopened = _reopen(config)
-    recovery = ArchiveTransferService(reopened, config=config)
-    # Whatever the crash interrupted is recoverable from evidence — never ambiguous.
-    # (A crash on the last barrier can land on a terminal item and leave none.)
-    verdicts = recovery.journal.classify_all()
-    assert all(v["classification"] in {RESUMABLE, FORWARD} for v in verdicts), verdicts
-
-    recovery.recover()
-    recovery.apply(plan["id"])  # finish whatever the crash never started
-
-    assert _contents(archive) == before
-    assert _contents(lib) == []
-    assert recovery.journal.blocks_mutation() is False
-    assert recovery.journal.plan_state(plan["id"]) == "complete"
-
-
-@pytest.mark.parametrize("barrier", [ArchiveState.VERIFIED, "source_removed"])
-def test_recovery_is_idempotent_across_repeated_restarts(tmp_path, barrier):
-    config, sf, lib, archive = _env(tmp_path)
-    _album(sf, lib)
-    service, plan = _plan(sf, config, archive)
-    _crash_during_apply(config, tmp_path, lib, plan, barrier)
-
-    reopened = _reopen(config)
-    recovery = ArchiveTransferService(reopened, config=config)
-    recovery.recover()
-    settled = (_contents(lib), _contents(archive), read_manifest(archive / "rome"))
-
-    for _ in range(2):
-        recovery.recover()
-        assert (_contents(lib), _contents(archive), read_manifest(archive / "rome")) == settled
-
-
-def test_a_crash_before_anything_was_published_leaves_the_source_intact(tmp_path):
-    config, sf, lib, archive = _env(tmp_path)
-    folder = _album(sf, lib, names=("a.jpg",))
-    service, plan = _plan(sf, config, archive)
-
-    _crash_during_apply(config, tmp_path, lib, plan, ArchiveState.TRANSFERRING)
-
-    recovery = ArchiveTransferService(_reopen(config), config=config)
-    verdict = recovery.journal.classify_all()[0]
-    assert verdict["classification"] == RESUMABLE
-    assert (folder / "a.jpg").exists()
-    assert not (archive / "rome" / "a.jpg").exists()
-
-    assert recovery.recover() == {"resumed": 1, "completed": 0, "manual": 0}
-    assert recovery.journal.operations(plan["id"])[0]["journal_state"] == ArchiveState.PLANNED
-
-
-def test_a_crash_after_the_source_was_removed_finishes_the_bookkeeping(tmp_path):
-    """The dangerous window: the original is gone and the database still points at
-    it. Recovery must complete the record, never re-transfer or report loss."""
-    config, sf, lib, archive = _env(tmp_path)
-    folder = _album(sf, lib, names=("a.jpg",))
-    service, plan = _plan(sf, config, archive)
-    expected = service.journal.operations(plan["id"])[0]["expected_sha256"]
-
-    _crash_during_apply(config, tmp_path, lib, plan, "source_removed")
-
-    reopened = _reopen(config)
-    recovery = ArchiveTransferService(reopened, config=config)
-    verdict = recovery.journal.classify_all()[0]
-    assert verdict["classification"] == FORWARD
-    assert not (folder / "a.jpg").exists()
-    assert sha256_file(archive / "rome" / "a.jpg") == expected
-    with reopened() as session:
-        stranded = session.scalar(select(Asset))
-        assert stranded.current_path is not None, "the crash happened before the DB update"
-
-    assert recovery.recover() == {"resumed": 0, "completed": 1, "manual": 0}
-
-    with reopened() as session:
-        asset = session.scalar(select(Asset))
-        assert asset.current_path is None
-        assert asset.availability_state == "archived_online"
-        assert asset.archive_path == "rome/a.jpg"
-    assert recovery.journal.operations(plan["id"])[0]["journal_state"] == ArchiveState.COMPLETE
-    assert len(read_manifest(archive / "rome")) == 1, "the manifest is not duplicated"
-
-
-def test_an_archive_copy_that_changed_after_the_crash_blocks_for_a_human(tmp_path):
-    """Wrong bytes at the destination can never justify deleting the original, and
-    are never overwritten either."""
-    config, sf, lib, archive = _env(tmp_path)
-    folder = _album(sf, lib, names=("a.jpg",))
-    service, plan = _plan(sf, config, archive)
-    _crash_during_apply(config, tmp_path, lib, plan, ArchiveState.VERIFIED, force_copy=True)
-    (archive / "rome" / "a.jpg").write_bytes(b"tampered with while the app was down")
-
-    recovery = ArchiveTransferService(_reopen(config), config=config)
-    verdict = recovery.journal.classify_all()[0]
-
-    assert verdict["classification"] == MANUAL
-    assert recovery.recover() == {"resumed": 0, "completed": 0, "manual": 1}
-    assert (folder / "a.jpg").exists(), "the source is kept while the archive is unproven"
-    assert (archive / "rome" / "a.jpg").read_bytes() == b"tampered with while the app was down"
-    # And the unresolved item keeps blocking further archiving.
-    assert recovery.journal.blocks_mutation() is True
-    report = ArchiveService(_reopen(config), config=config).preflight(plan["location_id"])
-    assert "archive_pending" in {issue["code"] for issue in report["blockers"]}
-
-
-def test_a_verified_archive_whose_copy_vanished_is_never_reported_as_archived(tmp_path):
-    config, sf, lib, archive = _env(tmp_path)
-    folder = _album(sf, lib, names=("a.jpg",))
-    service, plan = _plan(sf, config, archive)
-    _crash_during_apply(config, tmp_path, lib, plan, ArchiveState.REMOVING, force_copy=True)
-    (archive / "rome" / "a.jpg").unlink()  # the medium lost it
-
-    recovery = ArchiveTransferService(_reopen(config), config=config)
-
-    assert recovery.journal.classify_all()[0]["classification"] == MANUAL
-    assert recovery.recover()["manual"] == 1
-    assert (folder / "a.jpg").exists()
-    assert recovery.journal.operations(plan["id"])[0]["journal_state"] != ArchiveState.COMPLETE

tests/integration/test_archive_transfer.py

@@ -1,480 +0,0 @@
-"""Transferring, verifying, and removing active sources (US06-02).
-
-Every case here asks the same question the service exists to answer: could an
-original leave active storage without a durable, byte-identical archive copy? The
-files are real, the hashes are real, and each failure path asserts that the source
-is still exactly where it was.
-
-Cross-filesystem behaviour is forced by patching the device comparison rather than
-by requiring a second real filesystem in CI — the copy/verify/publish code that runs
-is the production one.
-"""
-
-import json
-import uuid
-from datetime import datetime, timezone
-
-import pytest
-from fastapi.testclient import TestClient
-from sqlalchemy import select
-
-from photo_pipeline.api.app import create_app
-from photo_pipeline.config import Config
-from photo_pipeline.db import create_db_engine, create_session_factory, run_migrations
-from photo_pipeline.models import Asset, AssetPath, UploadBatch, UploadItem
-from photo_pipeline.services import archive_transfer as transfer_module
-from photo_pipeline.services.archive_journal import ArchiveState
-from photo_pipeline.services.archive_transfer import (
-    MANIFEST_NAME,
-    ArchiveTransferService,
-    read_manifest,
-)
-from photo_pipeline.services.archives import ArchiveError, ArchiveService
-from photo_pipeline.services.hashing import sha256_file
-
-pytestmark = pytest.mark.phase_f  # part of the Phase F acceptance gate (US06-06)
-
-NOW = datetime(2026, 1, 1, tzinfo=timezone.utc)
-
-
-# ── environment ──────────────────────────────────────────────────────────────
-
-
-def _env(tmp_path):
-    (tmp_path / "data").mkdir(exist_ok=True)
-    lib = tmp_path / "lib"
-    lib.mkdir(exist_ok=True)
-    archive = tmp_path / "archive"
-    archive.mkdir(exist_ok=True)
-    config = Config.from_env(
-        {
-            "PHOTO_PIPELINE_DATA_DIR": str(tmp_path / "data"),
-            "PHOTO_PIPELINE_LIBRARY_ROOTS": str(lib),
-            "PHOTO_PIPELINE_ARCHIVE_FREE_SPACE_RESERVE_BYTES": "0",
-        }
-    )
-    run_migrations(config.database_url)
-    return config, create_session_factory(create_db_engine(config.database_url)), lib, archive
-
-
-def _album(sf, lib, album="rome", names=("a.jpg", "b.jpg")):
-    """A real album whose assets carry the verified upload evidence archiving needs."""
-    folder = lib / album
-    folder.mkdir(parents=True, exist_ok=True)
-    ids = []
-    with sf() as session:
-        batch_id = str(uuid.uuid4())
-        session.add(
-            UploadBatch(
-                id=batch_id,
-                album=album,
-                folder=str(folder),
-                album_name=album,
-                state="succeeded",
-                preflight_token="v1:test",
-                outcome_state="verified",
-                created_at=NOW,
-            )
-        )
-        for name in names:
-            path = folder / name
-            path.write_bytes(f"{album}/{name} bytes".encode() * 8)
-            asset_id = str(uuid.uuid4())
-            ids.append(asset_id)
-            session.add(
-                Asset(
-                    id=asset_id,
-                    original_path=str(path),
-                    current_path=str(path),
-                    discovered_at=NOW,
-                    hash_version=1,
-                    byte_size=path.stat().st_size,
-                    current_sha256=sha256_file(path),
-                )
-            )
-            session.add(AssetPath(asset_id=asset_id, path=str(path), valid_from=NOW))
-            session.add(
-                UploadItem(
-                    batch_id=batch_id,
-                    asset_id=asset_id,
-                    path=str(path),
-                    sha256=sha256_file(path),
-                    sha1="0" * 40,
-                    state="sent",
-                    outcome="uploaded",
-                )
-            )
-        session.commit()
-    return folder, ids
-
-
-def _location(sf, config, archive, name="external"):
-    return ArchiveService(sf, config=config).register(name, str(archive))
-
-
-def _plan(sf, config, location_id, albums=None):
-    service = ArchiveTransferService(sf, config=config)
-    token = ArchiveService(sf, config=config).preflight(location_id, albums)["token"]
-    return service, service.create(location_id, albums, token=token)
-
-
-def _contents(*roots):
-    """Every file's bytes under the given roots, ignoring our own bookkeeping."""
-    return sorted(
-        path.read_bytes()
-        for root in roots
-        for path in root.rglob("*")
-        if path.is_file() and path.name != MANIFEST_NAME and not path.name.startswith(".")
-    )
-
-
-def _states(service, plan_id):
-    return [row["journal_state"] for row in service.journal.operations(plan_id)]
-
-
-def _asset(sf, asset_id):
-    with sf() as session:
-        return session.get(Asset, asset_id)
-
-
-# ── planning ─────────────────────────────────────────────────────────────────
-
-
-def test_a_plan_records_every_file_with_its_expected_hash_and_moves_nothing(tmp_path):
-    config, sf, lib, archive = _env(tmp_path)
-    folder, ids = _album(sf, lib)
-    location = _location(sf, config, archive)
-    before = _contents(lib)
-
-    service, plan = _plan(sf, config, location["id"])
-
-    assert plan["state"] == "planned" and plan["asset_count"] == 2
-    assert sorted(op["asset_id"] for op in plan["operations"]) == sorted(ids)
-    for operation in plan["operations"]:
-        source = folder / operation["archive_path"].split("/")[-1]
-        assert operation["expected_sha256"] == sha256_file(source)
-        assert operation["archive_path"].startswith("rome/")
-        assert operation["journal_state"] == ArchiveState.PLANNED
-    assert _contents(lib) == before
-    assert _contents(archive) == []
-
-
-def test_a_stale_token_cannot_create_a_plan(tmp_path):
-    config, sf, lib, archive = _env(tmp_path)
-    folder, _ = _album(sf, lib)
-    location = _location(sf, config, archive)
-    token = ArchiveService(sf, config=config).preflight(location["id"])["token"]
-
-    (folder / "a.jpg").write_bytes(b"edited after approval")
-
-    with pytest.raises(ArchiveError) as error:
-        ArchiveTransferService(sf, config=config).create(location["id"], token=token)
-    assert error.value.code == "stale_token"
-
-
-def test_a_blocked_scope_cannot_create_a_plan(tmp_path):
-    """No verified upload means Immich may not hold these bytes; archiving would
-    remove the only copy."""
-    config, sf, lib, archive = _env(tmp_path)
-    folder = lib / "rome"
-    folder.mkdir()
-    path = folder / "a.jpg"
-    path.write_bytes(b"never uploaded")
-    with sf() as session:
-        session.add(
-            Asset(
-                id=str(uuid.uuid4()),
-                original_path=str(path),
-                current_path=str(path),
-                discovered_at=NOW,
-                hash_version=1,
-                byte_size=path.stat().st_size,
-                current_sha256=sha256_file(path),
-            )
-        )
-        session.commit()
-    location = _location(sf, config, archive)
-    token = ArchiveService(sf, config=config).preflight(location["id"])["token"]
-
-    with pytest.raises(ArchiveError) as error:
-        ArchiveTransferService(sf, config=config).create(location["id"], token=token)
-
-    assert error.value.code == "blocked"
-    assert path.exists()
-
-
-# ── the transfer ─────────────────────────────────────────────────────────────
-
-
-@pytest.mark.parametrize("same_filesystem", [True, False])
-def test_archiving_verifies_the_copy_before_the_source_is_removed(
-    tmp_path, monkeypatch, same_filesystem
-):
-    config, sf, lib, archive = _env(tmp_path)
-    folder, ids = _album(sf, lib)
-    location = _location(sf, config, archive)
-    service, plan = _plan(sf, config, location["id"])
-    expected = {op["asset_id"]: op["expected_sha256"] for op in plan["operations"]}
-    before = _contents(lib)
-    if not same_filesystem:
-        # Force the copy-verify-publish path without needing a second real volume.
-        monkeypatch.setattr(transfer_module, "_same_filesystem", lambda *_: False)
-
-    result = service.apply(plan["id"])
-
-    assert result == {
-        "plan_id": plan["id"],
-        "archived": 2,
-        "failed": 0,
-        "skipped": 0,
-        "state": "complete",
-    }
-    # The bytes moved: nothing is left in the library, everything is in the archive.
-    assert _contents(archive) == before
-    assert _contents(lib) == []
-    assert not folder.exists(), "an emptied album folder is not left behind"
-    for asset_id, digest in expected.items():
-        archived = archive / "rome" / _asset(sf, asset_id).archive_path.split("/")[-1]
-        assert sha256_file(archived) == digest
-    assert _states(service, plan["id"]) == [ArchiveState.COMPLETE] * 2
-    # No transfer temporaries survive either path.
-    assert not list((archive / "rome").glob(".archive-*"))
-
-
-@pytest.mark.parametrize("same_filesystem", [True, False])
-def test_the_manifest_on_the_medium_matches_the_archived_bytes(
-    tmp_path, monkeypatch, same_filesystem
-):
-    """The manifest is the medium's own record: it must be usable to verify the
-    archive with no database at all."""
-    config, sf, lib, archive = _env(tmp_path)
-    _album(sf, lib)
-    location = _location(sf, config, archive)
-    service, plan = _plan(sf, config, location["id"])
-    if not same_filesystem:
-        monkeypatch.setattr(transfer_module, "_same_filesystem", lambda *_: False)
-
-    service.apply(plan["id"])
-
-    entries = read_manifest(archive / "rome")
-    assert len(entries) == 2
-    for entry in entries:
-        archived = archive / entry["archive_path"]
-        assert sha256_file(archived) == entry["sha256"]
-        assert entry["byte_size"] == archived.stat().st_size
-        assert entry["media_id"] == location["media_id"]
-        assert entry["plan_id"] == plan["id"] and entry["album"] == "rome"
-
-
-def test_archived_assets_keep_their_identity_and_gain_their_new_location(tmp_path):
-    config, sf, lib, archive = _env(tmp_path)
-    folder, ids = _album(sf, lib)
-    location = _location(sf, config, archive)
-    service, plan = _plan(sf, config, location["id"])
-
-    service.apply(plan["id"])
-
-    with sf() as session:
-        for asset_id in ids:
-            asset = session.get(Asset, asset_id)
-            assert asset is not None, "archiving never deletes the record"
-            assert asset.current_path is None
-            assert asset.availability_state == "archived_online"
-            assert asset.archive_location_id == location["id"]
-            assert (archive / asset.archive_path).exists()
-            occurrences = session.scalars(
-                select(AssetPath).where(AssetPath.asset_id == asset_id)
-            ).all()
-            active = [row for row in occurrences if row.valid_until is None]
-            assert [row.path for row in active] == [str(archive / asset.archive_path)]
-            closed = [row for row in occurrences if row.valid_until is not None]
-            assert [row.path for row in closed] == [str(folder / asset.archive_path.split("/")[-1])]
-
-
-def test_applying_the_same_plan_again_changes_nothing(tmp_path):
-    config, sf, lib, archive = _env(tmp_path)
-    _album(sf, lib)
-    location = _location(sf, config, archive)
-    service, plan = _plan(sf, config, location["id"])
-    service.apply(plan["id"])
-    archived = _contents(archive)
-    manifest = read_manifest(archive / "rome")
-
-    again = service.apply(plan["id"])
-
-    assert again["skipped"] == 2 and again["archived"] == 0 and again["failed"] == 0
-    assert _contents(archive) == archived
-    assert read_manifest(archive / "rome") == manifest
-
-
-# ── unexpected changes stop the item ─────────────────────────────────────────
-
-
-def test_an_occupied_destination_is_never_overwritten(tmp_path):
-    config, sf, lib, archive = _env(tmp_path)
-    folder, _ = _album(sf, lib, names=("a.jpg",))
-    location = _location(sf, config, archive)
-    service, plan = _plan(sf, config, location["id"])
-    (archive / "rome").mkdir()
-    (archive / "rome" / "a.jpg").write_bytes(b"someone else's file")
-
-    result = service.apply(plan["id"])
-
-    assert result["failed"] == 1 and result["state"] == "failed"
-    assert (archive / "rome" / "a.jpg").read_bytes() == b"someone else's file"
-    assert (folder / "a.jpg").exists(), "the source must survive a refused transfer"
-    assert service.journal.operations(plan["id"])[0]["error_code"] == "destination_exists"
-
-
-def test_a_source_edited_after_planning_is_neither_archived_nor_removed(tmp_path):
-    config, sf, lib, archive = _env(tmp_path)
-    folder, _ = _album(sf, lib, names=("a.jpg",))
-    location = _location(sf, config, archive)
-    service, plan = _plan(sf, config, location["id"])
-
-    (folder / "a.jpg").write_bytes(b"edited between approval and apply")
-    result = service.apply(plan["id"])
-
-    assert result["failed"] == 1
-    assert (folder / "a.jpg").read_bytes() == b"edited between approval and apply"
-    assert not (archive / "rome" / "a.jpg").exists()
-    assert service.journal.operations(plan["id"])[0]["error_code"] == "source_changed"
-
-
-def test_a_copy_that_lands_with_the_wrong_bytes_is_not_published(tmp_path, monkeypatch):
-    """The read-back hash — not the fact that a write returned — is what proves the
-    archive copy."""
-    config, sf, lib, archive = _env(tmp_path)
-    folder, _ = _album(sf, lib, names=("a.jpg",))
-    location = _location(sf, config, archive)
-    service, plan = _plan(sf, config, location["id"])
-    monkeypatch.setattr(transfer_module, "_same_filesystem", lambda *_: False)
-    monkeypatch.setattr(
-        transfer_module.shutil,
-        "copyfileobj",
-        lambda src, dst, length=0: dst.write(b"corrupted in flight"),
-    )
-
-    result = service.apply(plan["id"])
-
-    assert result["failed"] == 1
-    assert service.journal.operations(plan["id"])[0]["error_code"] == "copy_mismatch"
-    assert (folder / "a.jpg").exists()
-    assert not (archive / "rome" / "a.jpg").exists()
-    assert not list((archive / "rome").glob(".archive-*")), "the failed copy is cleaned up"
-
-
-def test_a_source_replaced_by_a_symlink_is_refused(tmp_path):
-    config, sf, lib, archive = _env(tmp_path)
-    folder, _ = _album(sf, lib, names=("a.jpg",))
-    location = _location(sf, config, archive)
-    service, plan = _plan(sf, config, location["id"])
-    elsewhere = tmp_path / "elsewhere.jpg"
-    elsewhere.write_bytes(b"not a library file")
-    (folder / "a.jpg").unlink()
-    (folder / "a.jpg").symlink_to(elsewhere)
-
-    result = service.apply(plan["id"])
-
-    assert result["failed"] == 1
-    assert elsewhere.exists() and (folder / "a.jpg").is_symlink()
-    assert not (archive / "rome" / "a.jpg").exists()
-
-
-def test_one_failed_file_does_not_stop_the_rest_of_the_album(tmp_path):
-    config, sf, lib, archive = _env(tmp_path)
-    folder, _ = _album(sf, lib, names=("a.jpg", "b.jpg"))
-    location = _location(sf, config, archive)
-    service, plan = _plan(sf, config, location["id"])
-    (folder / "a.jpg").write_bytes(b"edited after approval")
-
-    result = service.apply(plan["id"])
-
-    assert (result["archived"], result["failed"]) == (1, 1)
-    assert (folder / "a.jpg").exists() and not (folder / "b.jpg").exists()
-    assert (archive / "rome" / "b.jpg").exists()
-    assert folder.exists(), "a folder that still holds an unarchived file stays"
-
-
-def test_an_unresolved_transfer_blocks_the_next_preflight(tmp_path, monkeypatch):
-    config, sf, lib, archive = _env(tmp_path)
-    _album(sf, lib, "rome", names=("a.jpg",))
-    _album(sf, lib, "paris", names=("c.jpg",))
-    location = _location(sf, config, archive)
-    service, plan = _plan(sf, config, location["id"], ["rome"])
-    # Leave one item stuck mid-transfer, as a killed process would.
-    service.journal.begin(
-        service.journal.operations(plan["id"])[0]["id"], worker_id="w", fencing_token=1
-    )
-
-    report = ArchiveService(sf, config=config).preflight(location["id"], ["paris"])
-
-    assert report["state"] == "blocked"
-    assert "archive_pending" in {issue["code"] for issue in report["blockers"]}
-
-
-# ── API surface ──────────────────────────────────────────────────────────────
-
-
-def test_api_creates_a_plan_and_queues_it_on_the_archiver_lane(tmp_path):
-    config, sf, lib, archive = _env(tmp_path)
-    _album(sf, lib)
-
-    with TestClient(create_app(config)) as client:
-        location = client.post(
-            "/api/v1/archive-locations", json={"name": "external", "root": str(archive)}
-        ).json()
-        token = client.post(
-            "/api/v1/archive-preflight", json={"location_id": location["id"]}
-        ).json()["token"]
-        created = client.post(
-            "/api/v1/archive-plans", json={"location_id": location["id"], "token": token}
-        )
-        plan_id = created.json()["id"]
-        fetched = client.get(f"/api/v1/archive-plans/{plan_id}")
-        applied = client.post(f"/api/v1/archive-plans/{plan_id}/apply")
-        listed = client.get("/api/v1/archive-plans")
-
-    assert created.status_code == 201 and created.json()["asset_count"] == 2
-    assert fetched.status_code == 200 and len(fetched.json()["operations"]) == 2
-    assert applied.status_code == 200 and applied.json()["job"]["state"] == "queued"
-    assert applied.json()["job"]["lock_key"] == "archive"
-    assert [row["id"] for row in listed.json()["plans"]] == [plan_id]
-    # Queuing alone must not have touched a single file.
-    assert _contents(lib) and _contents(archive) == []
-
-
-def test_api_refuses_a_stale_token_and_an_unknown_plan(tmp_path):
-    config, sf, lib, archive = _env(tmp_path)
-    _album(sf, lib)
-    location = _location(sf, config, archive)
-
-    with TestClient(create_app(config)) as client:
-        stale = client.post(
-            "/api/v1/archive-plans",
-            json={"location_id": location["id"], "token": "v1:not-the-real-token"},
-        )
-        unknown = client.get("/api/v1/archive-plans/nope")
-        unknown_apply = client.post("/api/v1/archive-plans/nope/apply")
-
-    assert stale.status_code == 409 and stale.json()["error"]["code"] == "stale_token"
-    assert unknown.status_code == 404 and unknown_apply.status_code == 404
-
-
-def test_api_reports_recovery_state_for_an_interrupted_transfer(tmp_path):
-    config, sf, lib, archive = _env(tmp_path)
-    _album(sf, lib, names=("a.jpg",))
-    location = _location(sf, config, archive)
-    service, plan = _plan(sf, config, location["id"])
-    service.journal.begin(
-        service.journal.operations(plan["id"])[0]["id"], worker_id="w", fencing_token=1
-    )
-
-    with TestClient(create_app(config)) as client:
-        status = client.get("/api/v1/archive-recovery").json()
-        resolved = client.post("/api/v1/archive-recovery/resolve").json()
-
-    assert status["blocks_mutation"] is True
-    assert status["operations"][0]["classification"] == "resumable"
-    assert resolved == {"resumed": 1, "completed": 0, "manual": 0}
-    assert json.loads(json.dumps(resolved))  # plain JSON, nothing exotic

tests/integration/test_inventory_reconcile.py

@@ -167,8 +167,7 @@ def test_missing_file_is_flagged_not_deleted(
     assets = assets_by_id(make_factory(db_url))
     assert missing_id in assets  # not pruned
     assert assets[missing_id].missing_at is not None
-    # Nothing explains the absence: this is not an offline archive medium (US06-03).
+    assert assets[missing_id].availability_state == "active"
-    assert assets[missing_id].availability_state == "missing_unexpected"
 
 
 def test_reappearing_file_clears_missing(
@@ -187,7 +186,6 @@ def test_reappearing_file_clears_missing(
     inventory.scan(lib)
     assets = assets_by_id(make_factory(db_url))
     assert assets[asset_id].missing_at is None
-    assert assets[asset_id].availability_state == "active"
 
 
 def test_identity_and_state_durable_across_restart(

tests/integration/test_offline_assets.py

@@ -1,377 +0,0 @@
-"""Offline identity and review evidence (US06-03).
-
-An archived photo is not gone: it keeps its identity, its hashes, and enough
-evidence to be recognised in a duplicate cluster while its medium sits in a
-drawer. Every case here archives a *real* album through the real transfer, then
-takes the medium away by removing its marker — the same thing the service sees
-when an external disk is unplugged — and asks whether the application still tells
-the truth about where the bytes are.
-
-The distinction that matters throughout: an unmounted medium is
-``archived_offline`` (expected, harmless), a mounted medium with a hole in it is
-``missing_unexpected`` (needs a human). Confusing the two is how an archive
-quietly loses a photo.
-"""
-
-import shutil
-import uuid
-from datetime import datetime, timezone
-
-import numpy as np
-import pytest
-from fastapi.testclient import TestClient
-from PIL import Image
-from sqlalchemy import select
-
-from photo_pipeline.api.app import create_app
-from photo_pipeline.config import Config
-from photo_pipeline.db import create_db_engine, create_session_factory, run_migrations
-from photo_pipeline.models import Asset, Thumbnail, UploadBatch, UploadItem
-from photo_pipeline.services.archive_transfer import ArchiveTransferService
-from photo_pipeline.services.archives import MARKER_NAME, ArchiveService
-from photo_pipeline.services.availability import (
-    ACTIVE,
-    ARCHIVED_OFFLINE,
-    ARCHIVED_ONLINE,
-    MISSING_UNEXPECTED,
-)
-from photo_pipeline.services.duplicates import ClusterState, DuplicateService, Method
-from photo_pipeline.services.hashing import sha256_file
-from photo_pipeline.services.inventory import InventoryService
-from photo_pipeline.services.thumbnails import PROTECTED_SIZE, ThumbnailService, ThumbnailUnavailable
-
-pytestmark = pytest.mark.phase_f  # part of the Phase F acceptance gate (US06-06)
-
-NOW = datetime(2026, 1, 1, tzinfo=timezone.utc)
-
-
-# ── environment ──────────────────────────────────────────────────────────────
-
-
-def _env(tmp_path):
-    (tmp_path / "data").mkdir(exist_ok=True)
-    lib = tmp_path / "lib"
-    lib.mkdir(exist_ok=True)
-    archive = tmp_path / "archive"
-    archive.mkdir(exist_ok=True)
-    config = Config.from_env(
-        {
-            "PHOTO_PIPELINE_DATA_DIR": str(tmp_path / "data"),
-            "PHOTO_PIPELINE_LIBRARY_ROOTS": str(lib),
-            "PHOTO_PIPELINE_ARCHIVE_FREE_SPACE_RESERVE_BYTES": "0",
-        }
-    )
-    run_migrations(config.database_url)
-    return config, create_session_factory(create_db_engine(config.database_url)), lib, archive
-
-
-def structured(path, seed, size=(256, 192)):
-    """A deterministic, decodable photo — previews and pHashes must be real."""
-    path.parent.mkdir(parents=True, exist_ok=True)
-    rng = np.random.default_rng(seed)
-    w, h = size
-    base = np.zeros((h, w, 3), dtype=np.uint8)
-    for _ in range(6):
-        x0 = int(rng.integers(0, w - 60))
-        y0 = int(rng.integers(0, h - 60))
-        base[y0 : y0 + 60, x0 : x0 + 60] = rng.integers(0, 256, 3)
-    grad = np.linspace(0, 120, w, dtype=np.uint8)
-    base[:, :, 0] = np.clip(base[:, :, 0].astype(int) + grad[None, :], 0, 255)
-    Image.fromarray(base).save(path, quality=95)
-    return path
-
-
-def resized_copy(src, dst, scale=0.5):
-    with Image.open(src) as image:
-        image.resize(
-            (int(image.width * scale), int(image.height * scale)), Image.LANCZOS
-        ).save(dst, quality=95)
-    return dst
-
-
-def _uploaded(sf, lib, album="rome", seeds=(1, 2)):
-    """A scanned album carrying the verified upload evidence archiving requires."""
-    folder = lib / album
-    for index, seed in enumerate(seeds):
-        structured(folder / f"{index}.jpg", seed)
-    result = InventoryService(sf).scan(lib)
-    with sf() as session:
-        batch_id = str(uuid.uuid4())
-        session.add(
-            UploadBatch(
-                id=batch_id,
-                album=album,
-                folder=str(folder),
-                album_name=album,
-                state="succeeded",
-                preflight_token="v1:test",
-                outcome_state="verified",
-                created_at=NOW,
-            )
-        )
-        for path, asset_id in result.asset_ids.items():
-            session.add(
-                UploadItem(
-                    batch_id=batch_id,
-                    asset_id=asset_id,
-                    path=path,
-                    sha256=sha256_file(path),
-                    sha1="0" * 40,
-                    state="sent",
-                    outcome="uploaded",
-                )
-            )
-        session.commit()
-    return folder, result.asset_ids
-
-
-def _archive(sf, config, archive, albums=None):
-    service = ArchiveService(sf, config=config)
-    location = service.register("external", str(archive))
-    token = service.preflight(location["id"], albums)["token"]
-    transfers = ArchiveTransferService(sf, config=config)
-    plan = transfers.create(location["id"], albums, token=token)
-    transfers.apply(plan["id"])
-    return location
-
-
-def _unmount(archive):
-    """Take the medium away the way a real one goes: its marker stops answering."""
-    (archive / MARKER_NAME).rename(archive / f"{MARKER_NAME}.away")
-
-
-def _remount(archive):
-    (archive / f"{MARKER_NAME}.away").rename(archive / MARKER_NAME)
-
-
-def _assets(sf):
-    with sf() as session:
-        return {asset.id: asset for asset in session.scalars(select(Asset))}
-
-
-# ── availability ─────────────────────────────────────────────────────────────
-
-
-def test_archived_assets_report_online_offline_and_missing(tmp_path):
-    config, sf, lib, archive = _env(tmp_path)
-    _, ids = _uploaded(sf, lib)
-    _archive(sf, config, archive)
-    inventory = InventoryService(sf)
-
-    states = {a.availability_state for a in _assets(sf).values()}
-    assert states == {ARCHIVED_ONLINE}
-
-    _unmount(archive)
-    inventory.scan(lib)  # the album folder is gone from the active roots
-    assert {a.availability_state for a in _assets(sf).values()} == {ARCHIVED_OFFLINE}
-    assert all(a.missing_at is None for a in _assets(sf).values())  # not "missing"
-
-    _remount(archive)
-    inventory.scan(lib)
-    assert {a.availability_state for a in _assets(sf).values()} == {ARCHIVED_ONLINE}
-
-    # Mounted medium, absent file: that is not an offline archive, it needs a human.
-    victim = sorted(ids.values())[0]
-    with sf() as session:
-        asset = session.get(Asset, victim)
-        (archive / asset.archive_path).unlink()
-    inventory.scan(lib)
-    assert _assets(sf)[victim].availability_state == MISSING_UNEXPECTED
-
-
-def test_scan_never_prunes_or_flags_offline_assets(tmp_path):
-    config, sf, lib, archive = _env(tmp_path)
-    _, ids = _uploaded(sf, lib)
-    _archive(sf, config, archive)
-    _unmount(archive)
-
-    before = _assets(sf)
-    result = InventoryService(sf).scan(lib)
-
-    assert result.counts.get("missing") is None
-    after = _assets(sf)
-    assert set(after) == set(before) == set(ids.values())
-    for asset in after.values():
-        assert asset.availability_state == ARCHIVED_OFFLINE
-        assert asset.current_sha256 and asset.pixel_sha256  # hashes retained
-        assert asset.archive_location_id and asset.archive_path
-
-
-def test_active_missing_file_is_missing_unexpected_not_offline(tmp_path):
-    config, sf, lib, _archive_root = _env(tmp_path)
-    structured(lib / "loose" / "a.jpg", 7)
-    ids = InventoryService(sf).scan(lib).asset_ids
-    asset_id = next(iter(ids.values()))
-
-    (lib / "loose" / "a.jpg").unlink()
-    InventoryService(sf).scan(lib)
-
-    asset = _assets(sf)[asset_id]
-    assert asset.availability_state == MISSING_UNEXPECTED
-    assert asset.missing_at is not None
-
-
-# ── deduplication against archived originals ─────────────────────────────────
-
-
-def test_exact_copy_of_offline_asset_links_to_archived_canonical(tmp_path):
-    config, sf, lib, archive = _env(tmp_path)
-    folder, ids = _uploaded(sf, lib, seeds=(1,))
-    archived_id = next(iter(ids.values()))
-    original = next(iter(ids))
-    kept = tmp_path / "kept.jpg"
-    shutil.copy2(original, kept)
-    _archive(sf, config, archive)
-    _unmount(archive)
-
-    # The same photo turns up again in the active library while the disk is away.
-    (lib / "inbox").mkdir(parents=True, exist_ok=True)
-    shutil.copy2(kept, lib / "inbox" / "again.jpg")
-    scan = InventoryService(sf).scan(lib)
-    new_id = scan.asset_ids[str(lib / "inbox" / "again.jpg")]
-    assert scan.occurrences[str(lib / "inbox" / "again.jpg")] == "copied"
-
-    clusters = DuplicateService(sf).detect().clusters
-    exact = [c for c in clusters if c["method"] == Method.EXACT.value]
-    assert len(exact) == 1
-    cluster = exact[0]
-    # Byte-identical: linked directly, and the archived original stays canonical.
-    assert cluster["state"] == ClusterState.DECIDED.value
-    assert cluster["canonical_asset_id"] == archived_id
-    assert _assets(sf)[new_id].canonical_asset_id == archived_id
-
-
-def test_fuzzy_copy_of_offline_asset_requires_review_and_names_the_medium(tmp_path):
-    config, sf, lib, archive = _env(tmp_path)
-    folder, ids = _uploaded(sf, lib, seeds=(1,))
-    archived_id = next(iter(ids.values()))
-    original = next(iter(ids))
-    variant_source = resized_copy(original, tmp_path / "small.jpg")
-    _archive(sf, config, archive)
-    _unmount(archive)
-
-    (lib / "inbox").mkdir(parents=True, exist_ok=True)
-    shutil.copy2(variant_source, lib / "inbox" / "small.jpg")
-    InventoryService(sf).scan(lib)
-
-    duplicates = DuplicateService(sf)
-    clusters = duplicates.detect().clusters
-    perceptual = [c for c in clusters if c["method"] == Method.PERCEPTUAL.value]
-    assert len(perceptual) == 1
-
-    detail = duplicates.get_cluster(perceptual[0]["id"])
-    assert detail["state"] == ClusterState.OPEN.value  # never auto-decided
-    assert detail["requires_confirmation"] is True
-    assert detail["mount_required"] == ["external"]  # full-resolution needs the disk
-    archived = next(m for m in detail["members"] if m["asset_id"] == archived_id)
-    assert archived["availability_state"] == ARCHIVED_OFFLINE
-    assert archived["current_path"] is None
-    assert archived["archive_path"] and archived["evidence"]["phash"]
-    # The retained preview is what makes the offline member reviewable at all.
-    assert archived["preview"] == {"state": "ready", "protected": True, "size": PROTECTED_SIZE}
-
-
-# ── protected review evidence ────────────────────────────────────────────────
-
-
-def test_protected_preview_survives_quota_and_serves_while_offline(tmp_path):
-    config, sf, lib, archive = _env(tmp_path)
-    _, ids = _uploaded(sf, lib, seeds=(1,))
-    asset_id = next(iter(ids.values()))
-    _archive(sf, config, archive)
-    _unmount(archive)
-
-    thumbnails = ThumbnailService(sf, config)
-    served = thumbnails.generate(asset_id, PROTECTED_SIZE)
-    assert served.exists()  # rendered before the original left, not from the medium
-
-    # An aggressive quota may empty the cache, but not this evidence.
-    tight = ThumbnailService(sf, config.model_copy(update={"thumbnail_cache_quota_bytes": 1}))
-    tight._enforce_quota()
-    assert served.exists()
-    with sf() as session:
-        row = session.scalar(select(Thumbnail).where(Thumbnail.asset_id == asset_id))
-        assert row.protected is True
-    assert thumbnails.evidence(asset_id)["state"] == "ready"
-
-
-def test_offline_asset_without_preview_reports_unavailable(tmp_path):
-    """No preview and no medium is an honest 409, never a wrong picture."""
-    config, sf, lib, archive = _env(tmp_path)
-    _, ids = _uploaded(sf, lib, seeds=(1,))
-    asset_id = next(iter(ids.values()))
-    _archive(sf, config, archive)
-    with sf() as session:
-        for row in session.scalars(select(Thumbnail).where(Thumbnail.asset_id == asset_id)):
-            session.delete(row)
-        session.commit()
-    _unmount(archive)
-
-    with pytest.raises(ThumbnailUnavailable):
-        ThumbnailService(sf, config).generate(asset_id, 256)
-
-    _remount(archive)  # mounted again: the archived original is readable
-    assert ThumbnailService(sf, config).generate(asset_id, 256).exists()
-
-
-def test_offline_asset_is_browsable_through_the_api(tmp_path):
-    """The browser sees an archived asset, its medium, and its preview — offline."""
-    config, sf, lib, archive = _env(tmp_path)
-    _, ids = _uploaded(sf, lib, seeds=(1,))
-    asset_id = next(iter(ids.values()))
-    _archive(sf, config, archive)
-    _unmount(archive)
-
-    with TestClient(create_app(config)) as client:
-        client.post("/api/v1/inventory/scan")
-        listed = client.get("/api/v1/inventory/assets", params={"availability": ARCHIVED_OFFLINE})
-        assert listed.status_code == 200
-        item = next(row for row in listed.json()["items"] if row["id"] == asset_id)
-        assert item["current_path"] is None
-        assert item["archive_path"] == "rome/0.jpg"
-        assert item["missing"] is False
-
-        # Searching by the archived path still finds it.
-        found = client.get("/api/v1/inventory/assets", params={"q": "rome"}).json()
-        assert [row["id"] for row in found["items"]] == [asset_id]
-
-        preview = client.get(f"/api/v1/assets/{asset_id}/thumbnail", params={"size": 1280})
-        assert preview.status_code == 200
-        assert preview.headers["content-type"] == "image/webp"
-
-
-def test_offline_state_is_stable_across_restart(tmp_path):
-    config, sf, lib, archive = _env(tmp_path)
-    _, ids = _uploaded(sf, lib, seeds=(1, 2))
-    _archive(sf, config, archive)
-    _unmount(archive)
-    InventoryService(sf).scan(lib)
-    before = {
-        asset_id: (
-            asset.availability_state,
-            asset.archive_path,
-            asset.current_sha256,
-            asset.phash,
-        )
-        for asset_id, asset in _assets(sf).items()
-    }
-
-    # Restart: a fresh engine and session factory against the same database.
-    restarted = create_session_factory(create_db_engine(config.database_url))
-    after = {
-        asset_id: (
-            asset.availability_state,
-            asset.archive_path,
-            asset.current_sha256,
-            asset.phash,
-        )
-        for asset_id, asset in _assets(restarted).items()
-    }
-    assert after == before
-    assert set(after) == set(ids.values())
-
-    # And the medium coming back is picked up by the restarted process.
-    _remount(archive)
-    assert ArchiveService(restarted, config=config).locations()[0]["state"] == "online"
-    assert {a.availability_state for a in _assets(restarted).values()} == {ARCHIVED_ONLINE}
-    assert ACTIVE not in {a.availability_state for a in _assets(restarted).values()}

tests/integration/test_restore.py

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

tests/story_traceability.json

@@ -126,23 +126,6 @@
     ],
     "US06-01": [
       "tests/integration/test_archive_preflight.py"
-    ],
-    "US06-02": [
-      "tests/unit/test_archive_journal_states.py",
-      "tests/integration/test_archive_transfer.py",
-      "tests/integration/test_archive_recovery.py"
-    ],
-    "US06-03": [
-      "tests/integration/test_offline_assets.py"
-    ],
-    "US06-04": [
-      "tests/integration/test_restore.py"
-    ],
-    "US06-05": [
-      "tests/e2e/test_archive_ui.py"
-    ],
-    "US06-06": [
-      "tests/e2e/test_phase_f_pipeline.py"
     ]
   }
 }

tests/unit/test_archive_journal_states.py

@@ -1,105 +0,0 @@
-"""Archive journal state machine and evidence table (US06-02).
-
-The evidence table decides whether an original may be deleted, so every
-combination of journal state and disk reality is asserted here as a pure function —
-no database, no files. A wrong cell in this table is data loss.
-"""
-
-import pytest
-
-from photo_pipeline.services.archive_journal import (
-    ALLOWED_TRANSITIONS,
-    FORWARD,
-    MANUAL,
-    RESUMABLE,
-    TERMINAL_STATES,
-    UNSAFE_STATES,
-    ArchiveState,
-    _classify,
-    can_transition,
-)
-
-pytestmark = pytest.mark.phase_f
-
-
-# ── state machine ────────────────────────────────────────────────────────────
-
-
-def test_the_happy_path_is_the_only_way_forward():
-    assert can_transition(ArchiveState.PLANNED, ArchiveState.TRANSFERRING)
-    assert can_transition(ArchiveState.TRANSFERRING, ArchiveState.VERIFIED)
-    assert can_transition(ArchiveState.VERIFIED, ArchiveState.REMOVING)
-    assert can_transition(ArchiveState.REMOVING, ArchiveState.COMPLETE)
-    # No shortcut may skip verification before a source is removed.
-    assert not can_transition(ArchiveState.TRANSFERRING, ArchiveState.REMOVING)
-    assert not can_transition(ArchiveState.PLANNED, ArchiveState.VERIFIED)
-    assert not can_transition(ArchiveState.VERIFIED, ArchiveState.COMPLETE)
-
-
-def test_removal_never_goes_backwards():
-    """Once the source may be gone, retrying the transfer would archive nothing and
-    could overwrite the copy that is now the only one."""
-    assert ALLOWED_TRANSITIONS[ArchiveState.REMOVING] == {
-        ArchiveState.COMPLETE,
-        ArchiveState.FAILED,
-    }
-    assert not can_transition(ArchiveState.REMOVING, ArchiveState.TRANSFERRING)
-    assert not can_transition(ArchiveState.REMOVING, ArchiveState.PLANNED)
-
-
-def test_complete_is_terminal_and_failed_can_be_retried():
-    assert ALLOWED_TRANSITIONS[ArchiveState.COMPLETE] == set()
-    assert TERMINAL_STATES == {ArchiveState.COMPLETE}
-    assert can_transition(ArchiveState.FAILED, ArchiveState.TRANSFERRING)
-    assert can_transition(ArchiveState.FAILED, ArchiveState.PLANNED)
-
-
-def test_every_state_that_can_touch_the_disk_is_marked_unsafe():
-    assert UNSAFE_STATES == {
-        ArchiveState.TRANSFERRING,
-        ArchiveState.VERIFIED,
-        ArchiveState.REMOVING,
-    }
-    assert ArchiveState.PLANNED not in UNSAFE_STATES
-
-
-# ── evidence table ───────────────────────────────────────────────────────────
-
-
-@pytest.mark.parametrize(
-    "state,source,destination,matches,expected",
-    [
-        # Nothing published yet: the source is still the only copy.
-        (ArchiveState.TRANSFERRING, True, False, False, RESUMABLE),
-        (ArchiveState.FAILED, True, False, False, RESUMABLE),
-        # The archive copy is durable and correct: finish the remaining steps.
-        (ArchiveState.TRANSFERRING, True, True, True, FORWARD),
-        (ArchiveState.TRANSFERRING, False, True, True, FORWARD),
-        (ArchiveState.VERIFIED, True, True, True, FORWARD),
-        (ArchiveState.REMOVING, False, True, True, FORWARD),
-        (ArchiveState.FAILED, True, True, True, FORWARD),
-        # Wrong bytes at the destination: never overwrite, never remove.
-        (ArchiveState.TRANSFERRING, True, True, False, MANUAL),
-        (ArchiveState.VERIFIED, True, True, False, MANUAL),
-        (ArchiveState.REMOVING, False, True, False, MANUAL),
-        # The journal claims an archived copy that is not there.
-        (ArchiveState.VERIFIED, True, False, False, MANUAL),
-        (ArchiveState.REMOVING, False, False, False, MANUAL),
-        # Neither copy exists — never silently accepted as success.
-        (ArchiveState.TRANSFERRING, False, False, False, MANUAL),
-        (ArchiveState.FAILED, False, False, False, MANUAL),
-    ],
-)
-def test_classification_of_every_evidence_combination(
-    state, source, destination, matches, expected
-):
-    classification, reason = _classify(state, source, destination, matches)
-
-    assert classification == expected, reason
-
-
-def test_a_source_that_is_gone_without_an_archive_copy_is_never_called_recoverable():
-    """The one combination that must always stop: the original left active storage
-    and nothing verifiable took its place."""
-    for state in (ArchiveState.TRANSFERRING, ArchiveState.VERIFIED, ArchiveState.REMOVING):
-        assert _classify(state, False, False, False)[0] == MANUAL