2026-08-09 20:10:59 +02:00
28 changed files with 1983 additions and 7 deletions
--- a/frontend/css/app.css
+++ b/frontend/css/app.css
@@ -116,3 +116,41 @@ tbody tr:focus { box-shadow: inset 3px 0 0 var(--accent); }
 .card { background: var(--surface); border: 1px solid var(--border); border-radius: var(--radius); padding: 14px; }
 a.link { color: var(--accent); text-decoration: none; }
 a.link:hover { text-decoration: underline; }
+
+/* ── Phase B views: workflow stepper, safety, library, analyze, stats ─────── */
+.stepper { display: grid; gap: 12px; }
+.stage { display: grid; gap: 6px; }
+.stage-head { display: flex; align-items: center; gap: 10px; }
+.stage-icon { font-size: 18px; width: 20px; text-align: center; }
+.stage-actions { display: flex; gap: 12px; align-items: center; margin-top: 4px; }
+.blocker { color: var(--warn); font-size: 13px; }
+.last-run { font-size: 12px; }
+
+.badge.complete { color: var(--ok); border-color: var(--ok); }
+.badge.ready { color: var(--muted); }
+.badge.attention { color: var(--warn); border-color: var(--warn); }
+.badge.blocked { color: var(--danger); border-color: var(--danger); }
+.badge.nsfw { color: var(--danger); border-color: var(--danger); }
+.badge.sfw { color: var(--ok); border-color: var(--ok); }
+
+.alert.job-running { border-color: var(--warn); color: var(--warn); background: rgba(255, 207, 91, 0.1); }
+
+.counts-grid { display: grid; grid-template-columns: repeat(auto-fit, minmax(120px, 1fr)); gap: 12px; margin: 8px 0 16px; }
+.stat { text-align: center; }
+.stat-value { font-size: 26px; font-weight: 700; }
+
+.tags { display: flex; flex-wrap: wrap; gap: 6px; }
+.lib-card { display: grid; gap: 6px; }
+.lib-card img { width: 100%; height: 140px; object-fit: cover; background: var(--surface-2); border-radius: 8px; }
+.lib-desc { font-size: 13px; }
+
+.activity-log {
+  background: var(--surface-2);
+  border: 1px solid var(--border);
+  border-radius: 8px;
+  padding: 10px;
+  font-size: 12px;
+  max-height: 240px;
+  overflow: auto;
+  white-space: pre-wrap;
+}
--- a/frontend/index.html
+++ b/frontend/index.html
@@ -10,8 +10,13 @@
    <header class="app-header">
      <span class="brand">Photo Pipeline</span>
      <nav aria-label="Primary">
+        <a href="#/workflow" data-nav="workflow">Workflow</a>
        <a href="#/inventory" data-nav="inventory">Inventory</a>
        <a href="#/duplicates" data-nav="duplicates">Duplicates</a>
+        <a href="#/safety" data-nav="safety">Safety</a>
+        <a href="#/library" data-nav="library">Library</a>
+        <a href="#/analyze" data-nav="analyze">Analyze</a>
+        <a href="#/stats" data-nav="stats">Stats</a>
      </nav>
    </header>
    <main id="app" aria-live="polite"><!-- views render here --></main>
--- a/frontend/js/api.js
+++ b/frontend/js/api.js
@@ -66,4 +66,17 @@ export const api = {
    request(`/jobs/${encodeURIComponent(id)}/events?after=${after}`, opts),
  jobEventsStreamUrl: (id, after = 0) =>
    `${BASE}/jobs/${encodeURIComponent(id)}/events/stream?after=${after}`,
+
+  // ── Workflow / safety / analysis / library ──────────────────────────────
+  workflow: (opts = {}) => request("/workflow", opts),
+  safetyQueue: (params = {}, opts = {}) =>
+    request("/safety/queue?" + new URLSearchParams(params).toString(), opts),
+  safetyDecide: (payload, opts = {}) =>
+    request("/safety/decisions", { method: "POST", body: JSON.stringify(payload), ...opts }),
+  safetyScoreJob: (opts = {}) => request("/safety/jobs", { method: "POST", ...opts }),
+  analysisCounts: (opts = {}) => request("/analysis/counts", opts),
+  analysisJob: (opts = {}) => request("/analysis/jobs", { method: "POST", ...opts }),
+  libraryAssets: (params = {}, opts = {}) =>
+    request("/library/assets?" + new URLSearchParams(params).toString(), opts),
+  libraryStats: (opts = {}) => request("/library/stats", opts),
 };
--- a/frontend/js/app.js
+++ b/frontend/js/app.js
@@ -1,5 +1,13 @@
 import { api } from "./api.js";
 import { navigate, onRouteChange, parseHash } from "./router.js";
+import {
+  renderAnalyze,
+  renderLibrary,
+  renderSafety,
+  renderStats,
+  renderWorkflow,
+  setRender,
+} from "./views.js";

 const root = document.getElementById("app");

@@ -345,11 +353,18 @@ async function applyDecision(cluster, decision, canonicalId) {
 function render() {
  const { path, params } = parseHash();
  const clusterMatch = path.match(/^\/duplicates\/(.+)$/);
-  if (path === "/inventory" || path === "/") renderInventory(params);
+  if (path === "/workflow" || path === "/") renderWorkflow(root, params);
+  else if (path === "/inventory") renderInventory(params);
  else if (path === "/duplicates") renderClusters(params);
  else if (clusterMatch) renderClusterDetail(clusterMatch[1]);
+  else if (path === "/safety") renderSafety(root, params);
+  else if (path === "/library") renderLibrary(root, params);
+  else if (path === "/analyze") renderAnalyze(root, params);
+  else if (path === "/stats") renderStats(root, params);
  else show(errorBanner("Unknown view"));
 }

+// Let views re-render the current route after a mutation.
+setRender(render);
 onRouteChange(render);
 render();
--- a/frontend/js/dom.js
+++ b/frontend/js/dom.js
@@ -0,0 +1,29 @@
+// Shared DOM helpers used by every view. `el` builds nodes without innerHTML
+// injection; the rest are the small conveniences the views repeat.
+export function el(tag, attrs = {}, ...children) {
+  const node = document.createElement(tag);
+  for (const [key, value] of Object.entries(attrs)) {
+    if (value == null || value === false) continue;
+    if (key === "class") node.className = value;
+    else if (key === "dataset") Object.assign(node.dataset, value);
+    else if (key.startsWith("on") && typeof value === "function")
+      node.addEventListener(key.slice(2).toLowerCase(), value);
+    else node.setAttribute(key, value);
+  }
+  for (const child of children.flat()) {
+    if (child == null || child === false) continue;
+    node.appendChild(typeof child === "string" ? document.createTextNode(child) : child);
+  }
+  return node;
+}
+
+export function errorBanner(message) {
+  return el("div", { class: "alert", role: "alert" }, message);
+}
+
+export function setActiveNav(view) {
+  document.querySelectorAll("nav a[data-nav]").forEach((a) => {
+    if (a.dataset.nav === view) a.setAttribute("aria-current", "page");
+    else a.removeAttribute("aria-current");
+  });
+}
--- a/frontend/js/router.js
+++ b/frontend/js/router.js
@@ -1,7 +1,7 @@
 // Hash router. URLs are stable and shareable: the view, its filters, and paging
 // all live in the hash, so a reload restores exactly what was on screen.
 export function parseHash() {
-  const raw = location.hash.slice(1) || "/inventory";
+  const raw = location.hash.slice(1) || "/workflow";
  const [path, queryString = ""] = raw.split("?");
  return { path, params: Object.fromEntries(new URLSearchParams(queryString)) };
 }
--- a/frontend/js/tests/unit.js
+++ b/frontend/js/tests/unit.js
@@ -51,7 +51,7 @@ async function run() {
    ok("router restores non-empty filters", params.q === "beach" && params.offset === "0");
    ok("router drops empty filters", !("empty" in params));
    location.hash = "";
-    ok("router defaults to /inventory", parseHash().path === "/inventory");
+    ok("router defaults to /workflow", parseHash().path === "/workflow");
  }

  // ── api errors + cancellation ────────────────────────────────────────────
--- a/frontend/js/views.js
+++ b/frontend/js/views.js
@@ -0,0 +1,363 @@
+// Phase B operational views: Workflow home, Safety review, Library, Analyze, Stats.
+// Each render function fills the app root from the JSON API — no server-rendered
+// HTML. Mutating actions are disabled while a job holds the library_write lock and
+// say why (concept §shared interaction rules: read-only browsing stays available).
+import { api } from "./api.js";
+import { el, errorBanner, setActiveNav } from "./dom.js";
+import { navigate } from "./router.js";
+import { subscribeJob } from "./events.js";
+
+// Status is never color-only: an icon + label accompany every colored badge.
+const STATUS_ICON = { complete: "✓", ready: "○", attention: "!", blocked: "●" };
+
+function jobBanner(activeJob) {
+  if (!activeJob) return null;
+  return el(
+    "div",
+    { class: "alert job-running", role: "status", "data-testid": "job-running" },
+    `A ${activeJob.job_type} job is ${activeJob.state}. Mutating actions are paused; browsing stays available.`
+  );
+}
+
+// ── Workflow home ────────────────────────────────────────────────────────────
+export async function renderWorkflow(root) {
+  setActiveNav("workflow");
+  let data;
+  try {
+    data = await api.workflow();
+  } catch (error) {
+    root.replaceChildren(errorBanner(`Failed to load workflow: ${error.message}`));
+    return;
+  }
+  const active = data.active_job;
+  const cards = data.stages.map((stage) => {
+    const action = stage.action || {};
+    // A stage whose action starts a job is disabled while any job runs.
+    const disabled = Boolean(active && action.job_type);
+    const button = action.label
+      ? el(
+          "button",
+          {
+            "data-testid": `action-${stage.key}`,
+            disabled: disabled ? "disabled" : false,
+            title: disabled ? "a job is already running" : false,
+            onclick: () => triggerStageAction(stage),
+          },
+          action.label
+        )
+      : null;
+    return el(
+      "div",
+      { class: `card stage stage-${stage.status}`, "data-testid": `stage-${stage.key}` },
+      el(
+        "div",
+        { class: "stage-head" },
+        el("span", { class: "stage-icon", "aria-hidden": "true" }, STATUS_ICON[stage.status] || "○"),
+        el("strong", {}, stage.title),
+        el("span", { class: `badge ${stage.status}`, "data-testid": `status-${stage.key}` }, stage.status_label)
+      ),
+      el("div", { class: "muted counts", "data-testid": `counts-${stage.key}` }, countsLine(stage.counts)),
+      stage.blocker ? el("div", { class: "blocker", "data-testid": `blocker-${stage.key}` }, `⚠ ${stage.blocker}`) : null,
+      el("div", { class: "muted last-run" }, stage.last_run ? `Last run: ${stage.last_run}` : "Not run yet"),
+      el(
+        "div",
+        { class: "stage-actions" },
+        button,
+        action.route ? el("a", { class: "link", href: action.route }, "Details →") : null
+      )
+    );
+  });
+  root.replaceChildren(
+    el("h1", {}, "Workflow"),
+    jobBanner(active),
+    el("div", { class: "stepper" }, ...cards)
+  );
+}
+
+async function triggerStageAction(stage) {
+  const action = stage.action || {};
+  if (action.job_type === "safety_score") await startJob(api.safetyScoreJob);
+  else if (action.job_type === "analysis") await startJob(api.analysisJob);
+  if (action.route) navigate(action.route.replace(/^#/, ""));
+}
+
+async function startJob(call) {
+  try {
+    await call();
+  } catch (error) {
+    if (error.status !== 409) alert(`Could not start job: ${error.message}`);
+  }
+}
+
+function countsLine(counts) {
+  return Object.entries(counts || {})
+    .map(([key, value]) => `${value} ${key}`)
+    .join(" · ");
+}
+
+// ── Safety review ────────────────────────────────────────────────────────────
+const SAFETY_FILTERS = ["undecided", "sfw", "nsfw", "deferred"];
+
+export async function renderSafety(root, params) {
+  setActiveNav("safety");
+  const state = params.state || "undecided";
+  let data;
+  try {
+    data = await api.safetyQueue({ state });
+  } catch (error) {
+    root.replaceChildren(errorBanner(`Failed to load safety queue: ${error.message}`));
+    return;
+  }
+
+  const tabs = el(
+    "div",
+    { class: "toolbar", role: "tablist" },
+    ...SAFETY_FILTERS.map((filter) =>
+      el(
+        "button",
+        {
+          role: "tab",
+          "data-testid": `filter-${filter}`,
+          "aria-selected": filter === state ? "true" : "false",
+          onclick: () => navigate("/safety", { state: filter }),
+        },
+        filter
+      )
+    )
+  );
+
+  const rows = data.items.map((item) =>
+    el(
+      "tr",
+      { "data-testid": "safety-row", "data-asset-id": item.asset_id },
+      el("td", {}, item.current_path || item.asset_id),
+      el("td", { class: "muted" }, item.score == null ? "—" : item.score.toFixed(3)),
+      el("td", { class: "muted" }, item.suggested || "—"),
+      el(
+        "td",
+        {},
+        el("span", { class: `badge ${item.decision || ""}` }, item.decision || "undecided"),
+        item.exif_verified ? el("span", { class: "badge sfw", title: "EXIF verified" }, "✓ exif") : null
+      ),
+      el(
+        "td",
+        { class: "decision-bar" },
+        decideButton(item, "sfw", "SFW"),
+        decideButton(item, "nsfw", "NSFW"),
+        decideButton(item, "deferred", "Defer")
+      )
+    )
+  );
+
+  root.replaceChildren(
+    el("h1", {}, "Safety review"),
+    tabs,
+    el(
+      "table",
+      { "aria-label": "Safety queue" },
+      el(
+        "thead",
+        {},
+        el("tr", {}, el("th", {}, "Path"), el("th", {}, "Score"), el("th", {}, "Suggested"), el("th", {}, "Decision"), el("th", {}, "Actions"))
+      ),
+      el("tbody", { "data-testid": "safety-body" }, ...rows)
+    ),
+    data.items.length ? null : el("p", { class: "muted" }, "Nothing in this queue.")
+  );
+}
+
+function decideButton(item, decision, label) {
+  return el(
+    "button",
+    {
+      "data-testid": `decide-${decision}`,
+      class: item.decision === decision ? "primary" : "",
+      onclick: async (event) => {
+        event.target.disabled = true;
+        try {
+          await api.safetyDecide({ asset_id: item.asset_id, decision });
+          render();
+        } catch (error) {
+          event.target.disabled = false;
+          alert(`Decision failed: ${error.message}`);
+        }
+      },
+    },
+    label
+  );
+}
+
+// ── Library ──────────────────────────────────────────────────────────────────
+export async function renderLibrary(root, params) {
+  setActiveNav("library");
+  const q = params.q || "";
+  const offset = Number(params.offset || 0);
+  const limit = 60;
+  let data;
+  try {
+    data = await api.libraryAssets({ q, offset, limit, sort: params.sort || "path" });
+  } catch (error) {
+    root.replaceChildren(errorBanner(`Failed to load library: ${error.message}`));
+    return;
+  }
+
+  const search = el("input", {
+    type: "search",
+    placeholder: "Search descriptions and tags…",
+    value: q,
+    "aria-label": "Search library",
+    onchange: (event) => navigate("/library", { q: event.target.value, offset: 0 }),
+  });
+
+  const cards = data.rows.map((row) =>
+    el(
+      "div",
+      { class: "card lib-card", "data-testid": "library-card" },
+      el("img", { src: api.thumbnailUrl(row.asset_id, 256), alt: row.description || row.current_path, loading: "lazy" }),
+      el("div", { class: "lib-desc" }, row.description || el("span", { class: "muted" }, "(not analyzed)")),
+      el("div", { class: "muted" }, row.album || ""),
+      el("div", { class: "tags" }, ...(row.tags || []).slice(0, 8).map((t) => el("span", { class: "badge" }, t)))
+    )
+  );
+
+  root.replaceChildren(
+    el("h1", {}, "Library"),
+    el("div", { class: "toolbar" }, search, el("span", { class: "muted", "data-testid": "library-total" }, `${data.total} photos`)),
+    data.rows.length ? el("div", { class: "cluster-grid" }, ...cards) : el("p", { class: "muted" }, "No matching photos."),
+    pager("/library", params, offset, limit, data.total)
+  );
+}
+
+// ── Analyze ──────────────────────────────────────────────────────────────────
+export async function renderAnalyze(root) {
+  setActiveNav("analyze");
+  let counts, workflow;
+  try {
+    [counts, workflow] = await Promise.all([api.analysisCounts(), api.workflow()]);
+  } catch (error) {
+    root.replaceChildren(errorBanner(`Failed to load analysis: ${error.message}`));
+    return;
+  }
+  const active = workflow.active_job;
+  const log = el("pre", { class: "activity-log", "data-testid": "analyze-log" });
+
+  const runButton = el(
+    "button",
+    {
+      class: "primary",
+      "data-testid": "run-analysis",
+      disabled: active || counts.eligible === 0 ? "disabled" : false,
+      title: active ? "a job is already running" : counts.eligible === 0 ? "no confirmed-SFW assets" : false,
+      onclick: async () => {
+        runButton.disabled = true;
+        try {
+          const job = await api.analysisJob();
+          log.textContent = `Started analysis job ${job.id}\n`;
+          subscribeJob(job.id, {
+            onEvent: (event) => (log.textContent += `${event.type}${event.message ? ": " + event.message : ""}\n`),
+            onDone: () => {
+              log.textContent += "done\n";
+              renderAnalyze(root);
+            },
+          });
+        } catch (error) {
+          runButton.disabled = false;
+          log.textContent = `Could not start: ${error.message}\n`;
+        }
+      },
+    },
+    "Analyze eligible SFW assets"
+  );
+
+  root.replaceChildren(
+    el("h1", {}, "Analyze"),
+    jobBanner(active),
+    el(
+      "div",
+      { class: "counts-grid", "data-testid": "analyze-counts" },
+      stat("Eligible (SFW)", counts.eligible),
+      stat("Analyzed", counts.analyzed),
+      stat("Pending", counts.pending),
+      stat("Errors", counts.error)
+    ),
+    el("div", { class: "stage-actions" }, runButton),
+    log
+  );
+}
+
+// ── Stats ────────────────────────────────────────────────────────────────────
+export async function renderStats(root) {
+  setActiveNav("stats");
+  let data;
+  try {
+    data = await api.libraryStats();
+  } catch (error) {
+    root.replaceChildren(errorBanner(`Failed to load stats: ${error.message}`));
+    return;
+  }
+  root.replaceChildren(
+    el("h1", {}, "Stats"),
+    el("div", { class: "counts-grid", "data-testid": "stats-status" }, ...Object.entries(data.status).map(([k, v]) => stat(k, v))),
+    facetBlock("Top tags", data.top_tags),
+    facetBlock("Settings", data.setting),
+    facetBlock("Years", data.years),
+    el(
+      "div",
+      { "data-testid": "stats-albums" },
+      el("h2", {}, "Albums"),
+      el(
+        "table",
+        { "aria-label": "Albums" },
+        el("thead", {}, el("tr", {}, el("th", {}, "Album"), el("th", {}, "Done"), el("th", {}, "Total"))),
+        el(
+          "tbody",
+          {},
+          ...data.albums.map((album) =>
+            el("tr", {}, el("td", {}, album.album), el("td", {}, String(album.done)), el("td", {}, String(album.total)))
+          )
+        )
+      )
+    )
+  );
+}
+
+// ── shared bits ───────────────────────────────────────────────────────────────
+function stat(label, value) {
+  return el("div", { class: "stat card" }, el("div", { class: "stat-value" }, String(value ?? 0)), el("div", { class: "muted" }, label));
+}
+
+function facetBlock(title, items) {
+  if (!items || !items.length) return null;
+  return el(
+    "div",
+    {},
+    el("h2", {}, title),
+    el("div", { class: "tags" }, ...items.map((item) => el("span", { class: "badge" }, `${item.value} (${item.count})`)))
+  );
+}
+
+function pager(path, params, offset, limit, total) {
+  const start = total === 0 ? 0 : offset + 1;
+  const end = Math.min(offset + limit, total);
+  return el(
+    "div",
+    { class: "pager" },
+    el(
+      "button",
+      { disabled: offset === 0 ? "disabled" : false, onclick: () => navigate(path, { ...params, offset: Math.max(offset - limit, 0) }) },
+      "Previous"
+    ),
+    el("span", { class: "muted" }, `${start}–${end} of ${total}`),
+    el(
+      "button",
+      { disabled: end >= total ? "disabled" : false, onclick: () => navigate(path, { ...params, offset: offset + limit }) },
+      "Next"
+    )
+  );
+}
+
+// Re-render the current route (used after a mutation). Set by app.js.
+let render = () => {};
+export function setRender(fn) {
+  render = fn;
+}
--- a/migrations/versions/0005_safety_analysis_workflow.py
+++ b/migrations/versions/0005_safety_analysis_workflow.py
@@ -0,0 +1,63 @@
+"""Safety reviews and content-analysis results (US02-06).
+
+Revision ID: 0005_safety_analysis
+Revises: 0004_durable_jobs
+Create Date: 2026-08-06
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = "0005_safety_analysis"
+down_revision = "0004_durable_jobs"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "safety_reviews",
+        sa.Column("id", sa.String(), primary_key=True),
+        sa.Column("asset_id", sa.String(), sa.ForeignKey("assets.id"), nullable=False),
+        sa.Column("score", sa.Float(), nullable=True),
+        sa.Column("decision", sa.String(), nullable=True),
+        sa.Column("prior_decision", sa.String(), nullable=True),
+        sa.Column("reviewer", sa.String(), nullable=True),
+        sa.Column("exif_verified_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("result_sha256", sa.String(), nullable=True),
+        sa.Column(
+            "created_at",
+            sa.DateTime(timezone=True),
+            nullable=False,
+            server_default=sa.text("CURRENT_TIMESTAMP"),
+        ),
+    )
+    op.create_index("ix_safety_reviews_asset_id", "safety_reviews", ["asset_id"])
+
+    op.create_table(
+        "analysis_results",
+        sa.Column("asset_id", sa.String(), sa.ForeignKey("assets.id"), primary_key=True),
+        sa.Column("status", sa.String(), nullable=False, server_default="pending"),
+        sa.Column("description", sa.String(), nullable=True),
+        sa.Column("tags", sa.String(), nullable=True),
+        sa.Column("people_count", sa.Integer(), nullable=True),
+        sa.Column("setting", sa.String(), nullable=True),
+        sa.Column("time_of_day", sa.String(), nullable=True),
+        sa.Column("season", sa.String(), nullable=True),
+        sa.Column("mood", sa.String(), nullable=True),
+        sa.Column("location_hint", sa.String(), nullable=True),
+        sa.Column("approx_year", sa.Integer(), nullable=True),
+        sa.Column("model", sa.String(), nullable=True),
+        sa.Column("prompt_version", sa.String(), nullable=True),
+        sa.Column("tokens_total", sa.Integer(), nullable=True),
+        sa.Column("raw_response", sa.String(), nullable=True),
+        sa.Column("error_message", sa.String(), nullable=True),
+        sa.Column("analyzed_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("exif_written_at", sa.DateTime(timezone=True), nullable=True),
+    )
+    op.create_index("ix_analysis_results_status", "analysis_results", ["status"])
+
+
+def downgrade() -> None:
+    op.drop_table("analysis_results")
+    op.drop_table("safety_reviews")
--- a/photo_pipeline/api/app.py
+++ b/photo_pipeline/api/app.py
@@ -14,7 +14,20 @@ from pathlib import Path
 from fastapi import FastAPI
 from fastapi.staticfiles import StaticFiles

-from photo_pipeline.api.routes import duplicates, health, inventory, jobs, thumbnails
+from photo_pipeline.api.routes import (
+    analysis,
+    duplicates,
+    health,
+    inventory,
+    jobs,
+    library,
+    safety,
+    thumbnails,
+    workflow,
+)
+
+# Registers the safety_score / analysis job handlers on import.
+import photo_pipeline.jobs.domain_handlers  # noqa: F401
 from photo_pipeline.config import Config
 from photo_pipeline.db import create_db_engine, create_session_factory, run_migrations
 from photo_pipeline.logging import configure_logging
@@ -46,6 +59,10 @@ def create_app(config: Config | None = None) -> FastAPI:
    app.include_router(duplicates.router, prefix="/api/v1")
    app.include_router(jobs.router, prefix="/api/v1")
    app.include_router(thumbnails.router, prefix="/api/v1")
+    app.include_router(workflow.router, prefix="/api/v1")
+    app.include_router(safety.router, prefix="/api/v1")
+    app.include_router(analysis.router, prefix="/api/v1")
+    app.include_router(library.router, prefix="/api/v1")
    # Static single-page app (hash-routed). Mounted last so /api/v1 wins.
    if FRONTEND_DIR.is_dir():
        app.mount("/app", StaticFiles(directory=FRONTEND_DIR, html=True), name="app")
--- a/photo_pipeline/api/routes/analysis.py
+++ b/photo_pipeline/api/routes/analysis.py
@@ -0,0 +1,50 @@
+"""Content-analysis API: results, counts, and the analysis job.
+
+The privacy gate lives in AnalysisService: enqueuing only ever targets confirmed-SFW
+assets, and the handler re-checks the gate per item, so an NSFW asset can never reach
+the provider even if its decision changes between enqueue and run.
+"""
+
+from __future__ import annotations
+
+from fastapi import APIRouter, Request
+from fastapi.responses import JSONResponse
+
+from photo_pipeline.jobs.domain_handlers import ANALYSIS, LIBRARY_WRITE_LOCK
+from photo_pipeline.services.analysis import AnalysisService
+from photo_pipeline.services.jobs import JobBlocked, JobService
+
+router = APIRouter(tags=["analysis"])
+
+
+def _service(request: Request) -> AnalysisService:
+    return AnalysisService(request.app.state.session_factory)
+
+
+def _error(status: int, code: str, message: str) -> JSONResponse:
+    return JSONResponse(status_code=status, content={"error": {"code": code, "message": message}})
+
+
+@router.get("/analysis/counts")
+def counts(request: Request) -> dict:
+    return _service(request).counts()
+
+
+@router.get("/analysis/results/{asset_id}")
+def result(asset_id: str, request: Request):
+    data = _service(request).get(asset_id)
+    if data is None:
+        return _error(404, "not_found", f"no analysis for {asset_id}")
+    return data
+
+
+@router.post("/analysis/jobs")
+def enqueue_analysis(request: Request):
+    ids = _service(request).eligible_asset_ids()
+    if not ids:
+        return _error(409, "nothing_eligible", "no confirmed-SFW assets ready for analysis")
+    jobs = JobService(request.app.state.session_factory)
+    try:
+        return jobs.enqueue(ANALYSIS, lock=LIBRARY_WRITE_LOCK, items=ids)
+    except JobBlocked as error:
+        return _error(409, error.code, str(error))
--- a/photo_pipeline/api/routes/library.py
+++ b/photo_pipeline/api/routes/library.py
@@ -0,0 +1,37 @@
+"""Library browsing and Stats API (read-only over analysis results)."""
+
+from __future__ import annotations
+
+from fastapi import APIRouter, Query, Request
+
+from photo_pipeline.services.library import LibraryService
+
+router = APIRouter(tags=["library"])
+
+_FILTER_KEYS = ("setting", "tod", "season", "status", "people", "year_min", "year_max", "has_location", "album")
+
+
+def _service(request: Request) -> LibraryService:
+    return LibraryService(request.app.state.session_factory)
+
+
+@router.get("/library/assets")
+def search(
+    request: Request,
+    q: str = "",
+    sort: str = "path",
+    limit: int = Query(60, ge=1, le=200),
+    offset: int = Query(0, ge=0),
+) -> dict:
+    filters = {key: value for key, value in request.query_params.items() if key in _FILTER_KEYS}
+    return _service(request).search(q=q, filters=filters, sort=sort, offset=offset, limit=limit)
+
+
+@router.get("/library/facets")
+def facets(request: Request) -> dict:
+    return _service(request).facets()
+
+
+@router.get("/library/stats")
+def stats(request: Request) -> dict:
+    return _service(request).stats()
--- a/photo_pipeline/api/routes/safety.py
+++ b/photo_pipeline/api/routes/safety.py
@@ -0,0 +1,60 @@
+"""Safety review, decisions, and scoring-job API.
+
+Decisions are synchronous commands (persist + EXIF checkpoint). Scoring is a durable
+job under the ``library_write`` lock, so only one mutating job runs at a time.
+"""
+
+from __future__ import annotations
+
+from fastapi import APIRouter, Query, Request
+from fastapi.responses import JSONResponse
+
+from photo_pipeline.jobs.domain_handlers import LIBRARY_WRITE_LOCK, SAFETY_SCORE
+from photo_pipeline.schemas import SafetyDecisionRequest
+from photo_pipeline.services.jobs import JobBlocked, JobService
+from photo_pipeline.services.safety import SafetyError, SafetyService
+
+router = APIRouter(tags=["safety"])
+
+
+def _service(request: Request) -> SafetyService:
+    return SafetyService(request.app.state.session_factory)
+
+
+def _error(status: int, code: str, message: str) -> JSONResponse:
+    return JSONResponse(status_code=status, content={"error": {"code": code, "message": message}})
+
+
+@router.get("/safety/queue")
+def review_queue(
+    request: Request,
+    state: str = "",
+    limit: int = Query(100, ge=1, le=500),
+    offset: int = Query(0, ge=0),
+) -> dict:
+    return _service(request).review_queue(state=state, limit=limit, offset=offset)
+
+
+@router.get("/safety/counts")
+def counts(request: Request) -> dict:
+    return _service(request).counts()
+
+
+@router.post("/safety/decisions")
+def decide(body: SafetyDecisionRequest, request: Request):
+    try:
+        return _service(request).decide(body.asset_id, body.decision)
+    except SafetyError as error:
+        return _error(404, "not_found", str(error))
+
+
+@router.post("/safety/jobs")
+def enqueue_scoring(request: Request):
+    ids = _service(request).scorable_asset_ids()
+    if not ids:
+        return _error(409, "nothing_to_score", "no eligible assets to score")
+    jobs = JobService(request.app.state.session_factory)
+    try:
+        return jobs.enqueue(SAFETY_SCORE, lock=LIBRARY_WRITE_LOCK, items=ids)
+    except JobBlocked as error:
+        return _error(409, error.code, str(error))
--- a/photo_pipeline/api/routes/workflow.py
+++ b/photo_pipeline/api/routes/workflow.py
@@ -0,0 +1,14 @@
+"""Workflow readiness API — powers the Workflow home stepper."""
+
+from __future__ import annotations
+
+from fastapi import APIRouter, Request
+
+from photo_pipeline.services.workflow import WorkflowService
+
+router = APIRouter(tags=["workflow"])
+
+
+@router.get("/workflow")
+def workflow(request: Request) -> dict:
+    return WorkflowService(request.app.state.session_factory).readiness()
--- a/photo_pipeline/jobs/domain_handlers.py
+++ b/photo_pipeline/jobs/domain_handlers.py
@@ -0,0 +1,36 @@
+"""Domain job handlers: safety scoring and content analysis (US02-06).
+
+Importing this module registers the ``safety_score`` and ``analysis`` job types so
+the generic worker can run them per item (one item = one ``asset_id``). Each handler
+delegates to its service, which owns the real work and the privacy gate. Handlers
+are idempotent: re-scoring or re-analyzing one asset is safe after an interrupted
+attempt.
+
+Providers/models are the service defaults here (real NsfwModel / vision provider);
+tests exercise the services directly with injected fakes rather than the worker.
+"""
+
+from __future__ import annotations
+
+from photo_pipeline.jobs.handlers import JobContext, register
+
+SAFETY_SCORE = "safety_score"
+ANALYSIS = "analysis"
+# Both mutate the library's metadata/derived state; one at a time (concept §one job).
+LIBRARY_WRITE_LOCK = "library_write"
+
+
+def _safety_score_item(asset_id: str, ctx: JobContext) -> None:
+    from photo_pipeline.services.safety import SafetyService
+
+    SafetyService(ctx.session_factory).score_assets([asset_id])
+
+
+def _analysis_item(asset_id: str, ctx: JobContext) -> None:
+    from photo_pipeline.services.analysis import AnalysisService
+
+    AnalysisService(ctx.session_factory).run([asset_id])
+
+
+register(SAFETY_SCORE, _safety_score_item)
+register(ANALYSIS, _analysis_item)
--- a/photo_pipeline/jobs/handlers.py
+++ b/photo_pipeline/jobs/handlers.py
@@ -26,12 +26,14 @@ class Cancelled(Exception):

@dataclass
 class JobContext:
-    """What a handler needs from the coordinator: cancellation checks and heartbeats."""
+    """What a handler needs from the coordinator: cancellation checks, heartbeats,
+    and the session factory to build domain services against."""

    job_id: str
    worker_id: str
    fencing_token: int
    service: "JobService"
+    session_factory: object | None = None

    def cancelled(self) -> bool:
        from photo_pipeline.services.jobs import JobState
--- a/photo_pipeline/jobs/worker.py
+++ b/photo_pipeline/jobs/worker.py
@@ -51,7 +51,7 @@ class Worker:

    def _process(self, job_id: str, job_type: str, token: int) -> None:
        handler = self.handlers[job_type]
-        ctx = JobContext(job_id, self.worker_id, token, self.service)
+        ctx = JobContext(job_id, self.worker_id, token, self.service, self._session_factory)
        self._reset_interrupted_items(job_id, token)

        cancelled = False
--- a/photo_pipeline/models/init.py
+++ b/photo_pipeline/models/init.py
@@ -12,6 +12,7 @@ from photo_pipeline.models.duplicates import (
 )
 from photo_pipeline.models.jobs import Job, JobEvent, JobItem
 from photo_pipeline.models.thumbnails import Thumbnail
+from photo_pipeline.models.workflow import AnalysisResult, SafetyReview

 __all__ = [
    "Asset",
@@ -23,4 +24,6 @@ __all__ = [
    "JobItem",
    "JobEvent",
    "Thumbnail",
+    "SafetyReview",
+    "AnalysisResult",
 ]
--- a/photo_pipeline/models/workflow.py
+++ b/photo_pipeline/models/workflow.py
@@ -0,0 +1,66 @@
+"""Safety review and content-analysis persistence (US02-06).
+
+Both tables key on the stable ``asset_id``, not a path — the donors used a
+path-keyed ``nsfw_scores.csv`` and a path-keyed ``photos`` table, which broke on
+every move/rename. ``safety_reviews`` is append-only history; the latest row per
+asset is the current decision (the privacy gate reads it). ``analysis_results`` is
+one current row per asset (the donor photos schema re-keyed to asset identity).
+
+Per-stage ``asset_stage_states``/``exif_projections`` from the concept are not
+modelled here: the workflow view derives its counts directly from these source
+tables plus duplicate clusters, which is enough for this story's gates.
+ponytail: add the full stage-state projection when a stage needs history the
+source tables can't reconstruct.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime
+
+from sqlalchemy import DateTime, ForeignKey, Integer, String, func
+from sqlalchemy.orm import Mapped, mapped_column
+
+from photo_pipeline.db import Base
+
+
+class SafetyReview(Base):
+    __tablename__ = "safety_reviews"
+
+    id: Mapped[str] = mapped_column(String, primary_key=True)
+    asset_id: Mapped[str] = mapped_column(ForeignKey("assets.id"), nullable=False, index=True)
+    # score without decision = scored-but-unreviewed; decision without score = manual.
+    score: Mapped[float | None] = mapped_column()
+    decision: Mapped[str | None] = mapped_column(String)  # sfw | nsfw | deferred
+    prior_decision: Mapped[str | None] = mapped_column(String)
+    reviewer: Mapped[str | None] = mapped_column(String)
+    # Set once the mutually-exclusive safety keyword is written to EXIF and read
+    # back — upload eligibility depends on this verified checkpoint.
+    exif_verified_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+    result_sha256: Mapped[str | None] = mapped_column(String)
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, server_default=func.now()
+    )
+
+
+class AnalysisResult(Base):
+    __tablename__ = "analysis_results"
+
+    asset_id: Mapped[str] = mapped_column(ForeignKey("assets.id"), primary_key=True)
+    # pending | analyzed | error | skipped_nsfw
+    status: Mapped[str] = mapped_column(String, nullable=False, default="pending")
+    description: Mapped[str | None] = mapped_column(String)
+    tags: Mapped[str | None] = mapped_column(String)  # JSON array string (donor shape)
+    people_count: Mapped[int | None] = mapped_column(Integer)
+    setting: Mapped[str | None] = mapped_column(String)
+    time_of_day: Mapped[str | None] = mapped_column(String)
+    season: Mapped[str | None] = mapped_column(String)
+    mood: Mapped[str | None] = mapped_column(String)
+    location_hint: Mapped[str | None] = mapped_column(String)
+    approx_year: Mapped[int | None] = mapped_column(Integer)
+    model: Mapped[str | None] = mapped_column(String)
+    prompt_version: Mapped[str | None] = mapped_column(String)
+    tokens_total: Mapped[int | None] = mapped_column(Integer)
+    raw_response: Mapped[str | None] = mapped_column(String)
+    error_message: Mapped[str | None] = mapped_column(String)
+    analyzed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+    exif_written_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
--- a/photo_pipeline/schemas/init.py
+++ b/photo_pipeline/schemas/init.py
@@ -2,5 +2,6 @@

 from photo_pipeline.schemas.duplicates import DecisionRequest
 from photo_pipeline.schemas.jobs import JobStartRequest
+from photo_pipeline.schemas.safety import SafetyDecisionRequest

-__all__ = ["DecisionRequest", "JobStartRequest"]
+__all__ = ["DecisionRequest", "JobStartRequest", "SafetyDecisionRequest"]
--- a/photo_pipeline/schemas/safety.py
+++ b/photo_pipeline/schemas/safety.py
@@ -0,0 +1,12 @@
+"""Safety API request contracts."""
+
+from __future__ import annotations
+
+from typing import Literal
+
+from pydantic import BaseModel
+
+
+class SafetyDecisionRequest(BaseModel):
+    asset_id: str
+    decision: Literal["sfw", "nsfw", "deferred"]
--- a/photo_pipeline/services/analysis.py
+++ b/photo_pipeline/services/analysis.py
@@ -0,0 +1,291 @@
+"""AnalysisService — content analysis scoped to confirmed-SFW assets.
+
+The privacy invariant is enforced here and nowhere else: the vision provider is
+called **only** for canonical assets whose latest safety decision is ``sfw``.
+Confirmed ``nsfw`` and undecided assets never reach the provider — ``run`` records
+them as ``skipped_nsfw``/leaves them pending without constructing a request. The
+provider is an injected adapter so tests assert this with a call-recording fake and
+no network/key.
+
+Extracted from photo_analyzer.analyze_image (the OpenAI-compatible Gemini call,
+result fields, and album-hint prompt) and its ``photos`` schema, re-keyed to
+``asset_id`` (donor_ledger.yaml: pa-analyze, pa-schema). Retry/rate-limit/throttle
+bookkeeping from the donor is out of scope for this story.
+ponytail: port the donor's retry+RPD throttling when analysis runs at real volume.
+"""
+
+from __future__ import annotations
+
+import json
+import uuid
+from datetime import datetime, timezone
+from typing import Protocol
+
+from sqlalchemy import func, select
+from sqlalchemy.orm import sessionmaker
+
+from photo_pipeline.integrations import exiftool
+from photo_pipeline.models import AnalysisResult, Asset, SafetyReview
+from photo_pipeline.services import hashing
+from photo_pipeline.services.safety import SFW
+
+MODEL = "gemini-2.5-flash"
+PROMPT_VERSION = "1"
+RESULT_FIELDS = (
+    "description",
+    "tags",
+    "people_count",
+    "setting",
+    "time_of_day",
+    "season",
+    "mood",
+    "location_hint",
+    "approx_year",
+)
+
+
+class VisionProvider(Protocol):
+    def analyze(self, path: str, *, album_hint: str) -> dict:
+        """Return the analysis fields for one image. Raises on unrecoverable error."""
+
+
+class AnalysisError(Exception):
+    pass
+
+
+def _now() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+class AnalysisService:
+    def __init__(self, session_factory: sessionmaker, *, provider: VisionProvider | None = None) -> None:
+        self._session_factory = session_factory
+        self._provider = provider
+
+    def _sfw_asset_ids(self, session) -> set[str]:
+        """Asset ids whose latest safety decision is ``sfw`` — the ONLY assets that
+        may reach the provider."""
+        latest: dict[str, str | None] = {}
+        for review in session.scalars(select(SafetyReview).order_by(SafetyReview.created_at)):
+            latest[review.asset_id] = review.decision
+        return {aid for aid, decision in latest.items() if decision == SFW}
+
+    def eligible_asset_ids(self) -> list[str]:
+        """Confirmed-SFW canonical active assets without a completed analysis."""
+        with self._session_factory() as session:
+            sfw = self._sfw_asset_ids(session)
+            if not sfw:
+                return []
+            assets = session.scalars(
+                select(Asset).where(
+                    Asset.id.in_(sfw),
+                    Asset.canonical_asset_id.is_(None),
+                    Asset.availability_state == "active",
+                    Asset.current_path.is_not(None),
+                )
+            )
+            done = set(
+                session.scalars(
+                    select(AnalysisResult.asset_id).where(AnalysisResult.status == "analyzed")
+                )
+            )
+            return [a.id for a in assets if a.id not in done]
+
+    def counts(self) -> dict[str, int]:
+        with self._session_factory() as session:
+            sfw = self._sfw_asset_ids(session)
+            rows = dict(
+                session.execute(
+                    select(AnalysisResult.status, func.count()).group_by(AnalysisResult.status)
+                ).all()
+            )
+            analyzed = int(rows.get("analyzed", 0))
+            errored = int(rows.get("error", 0))
+            return {
+                "eligible": len(sfw),
+                "analyzed": analyzed,
+                "error": errored,
+                "pending": max(len(sfw) - analyzed - errored, 0),
+            }
+
+    def run(self, asset_ids: list[str] | None = None) -> dict:
+        """Analyze the given assets (default: all eligible). Enforces the gate.
+
+        Returns ``{analyzed, skipped, errors}``. ``skipped`` counts assets that were
+        requested but are not confirmed SFW — the provider is never called for them.
+        """
+        with self._session_factory() as session:
+            sfw = self._sfw_asset_ids(session)
+            paths = {
+                a.id: a.current_path
+                for a in session.scalars(select(Asset).where(Asset.canonical_asset_id.is_(None)))
+                if a.current_path
+            }
+        requested = asset_ids if asset_ids is not None else self.eligible_asset_ids()
+        provider = self._provider or _default_provider()
+
+        analyzed = skipped = errors = 0
+        for asset_id in requested:
+            if asset_id not in sfw:
+                # Gate: not confirmed SFW → never construct a provider request.
+                self._store(asset_id, status="skipped_nsfw", result=None, error=None, tokens=0, raw="")
+                skipped += 1
+                continue
+            path = paths.get(asset_id)
+            if not path:
+                skipped += 1
+                continue
+            try:
+                result = provider.analyze(path, album_hint=_album_hint(path))
+            except Exception as error:  # provider/validation failure is per-asset
+                self._store(asset_id, status="error", result=None, error=str(error), tokens=0, raw="")
+                errors += 1
+                continue
+            self._store(
+                asset_id,
+                status="analyzed",
+                result=result,
+                error=None,
+                tokens=int(result.get("_tokens", 0)) if isinstance(result, dict) else 0,
+                raw=json.dumps(result, ensure_ascii=False),
+            )
+            _write_analysis_exif(path, result)
+            analyzed += 1
+        return {"analyzed": analyzed, "skipped": skipped, "errors": errors}
+
+    def _store(self, asset_id, *, status, result, error, tokens, raw) -> None:
+        now = _now()
+        with self._session_factory() as session:
+            row = session.get(AnalysisResult, asset_id) or AnalysisResult(asset_id=asset_id)
+            row.status = status
+            row.error_message = error
+            row.model = MODEL
+            row.prompt_version = PROMPT_VERSION
+            row.tokens_total = tokens
+            row.raw_response = raw or None
+            if status == "analyzed" and isinstance(result, dict):
+                row.description = result.get("description")
+                row.tags = json.dumps(result.get("tags", []), ensure_ascii=False)
+                row.people_count = result.get("people_count")
+                row.setting = result.get("setting")
+                row.time_of_day = result.get("time_of_day")
+                row.season = result.get("season")
+                row.mood = result.get("mood")
+                row.location_hint = result.get("location_hint")
+                row.approx_year = result.get("approx_year")
+                row.analyzed_at = now
+                row.exif_written_at = now
+            session.add(row)
+            session.commit()
+
+    def get(self, asset_id: str) -> dict | None:
+        with self._session_factory() as session:
+            row = session.get(AnalysisResult, asset_id)
+            return _result_dict(row) if row else None
+
+
+def _album_hint(path: str) -> str:
+    from pathlib import Path
+
+    return Path(path).parent.name
+
+
+def _write_analysis_exif(path: str, result: dict) -> None:
+    """Additive analysis keywords into EXIF (Keywords/Subject), preserving safety
+    and user keywords. The donor also wrote a managed caption; only keywords are
+    written here via the shared adapter.
+    ponytail: add the managed ``AI:`` caption segment + read-back preservation check
+    when the analysis EXIF checkpoint is hardened."""
+    tags = result.get("tags") if isinstance(result, dict) else None
+    if tags:
+        exiftool.apply_keywords(path, add=[str(t) for t in tags])
+
+
+def _result_dict(row: AnalysisResult) -> dict:
+    data = {field: getattr(row, field) for field in RESULT_FIELDS}
+    data["tags"] = json.loads(row.tags) if row.tags else []
+    data.update(
+        asset_id=row.asset_id,
+        status=row.status,
+        model=row.model,
+        prompt_version=row.prompt_version,
+        tokens_total=row.tokens_total,
+        error_message=row.error_message,
+    )
+    return data
+
+
+def _default_provider() -> VisionProvider:
+    return OpenAIVisionProvider()
+
+
+class OpenAIVisionProvider:
+    """The real provider: an OpenAI-compatible vision call (Gemini by default).
+
+    Extracted from photo_analyzer.analyze_image. Constructed lazily from env
+    (``OPENAI_API_KEY`` / ``OPENAI_BASE_URL``); never used in tests, which inject a
+    fake. Kept intentionally thin — no retry/throttle bookkeeping (see module note).
+    """
+
+    def __init__(self, *, model: str = MODEL, client=None) -> None:
+        self._model = model
+        self._client = client
+
+    def _ensure_client(self):
+        if self._client is None:
+            from openai import OpenAI
+
+            self._client = OpenAI()
+        return self._client
+
+    def analyze(self, path: str, *, album_hint: str) -> dict:
+        b64, mime = _prepare_image(path)
+        prompt = ANALYSIS_PROMPT + (
+            f'\n\nAlbum hint: this photo is filed in a folder named "{album_hint}". '
+            f'Folder names often contain the place and/or year — use it to inform '
+            f'"location_hint" and "approx_year", but trust the image if they conflict.'
+        )
+        response = self._ensure_client().chat.completions.create(
+            model=self._model,
+            max_tokens=4096,
+            messages=[
+                {
+                    "role": "user",
+                    "content": [
+                        {"type": "image_url", "image_url": {"url": f"data:{mime};base64,{b64}"}},
+                        {"type": "text", "text": prompt},
+                    ],
+                }
+            ],
+        )
+        raw = (response.choices[0].message.content or "").strip()
+        if raw.startswith("```"):
+            raw = raw.split("```")[1]
+            raw = raw[4:] if raw.startswith("json") else raw
+            raw = raw.strip()
+        result = json.loads(raw)
+        usage = getattr(response, "usage", None)
+        result["_tokens"] = usage.total_tokens if usage else 0
+        return result
+
+
+ANALYSIS_PROMPT = (
+    "Analyze this photograph and return ONLY a JSON object with keys: description "
+    "(one clear sentence), tags (8-12 specific keywords), people_count (integer), "
+    "setting, time_of_day, season, mood, location_hint (or null), approx_year "
+    "(integer or null)."
+)
+
+
+def _prepare_image(path: str) -> tuple[str, str]:
+    import base64
+    from io import BytesIO
+
+    from PIL import Image
+
+    with Image.open(path) as image:
+        image = image.convert("RGB")
+        image.thumbnail((1024, 1024))
+        buffer = BytesIO()
+        image.save(buffer, format="JPEG", quality=85)
+    return base64.b64encode(buffer.getvalue()).decode(), "image/jpeg"
--- a/photo_pipeline/services/library.py
+++ b/photo_pipeline/services/library.py
@@ -0,0 +1,186 @@
+"""LibraryService — read-only Library and Stats over analysis results.
+
+Ports webapp/query.py (search, facets, stats, top-tag/people/year aggregates) onto
+the shared database, re-keyed to ``asset_id`` and joined to ``assets`` for the
+current path. The donor read a path-keyed ``photos`` table with an FTS5 index; here
+search is a tokenised ``LIKE`` over description + tags.
+ponytail: restore FTS5 (rank-ordered relevance) if search recall/latency matters at
+real library size — LIKE is fine for browsing tens of thousands of rows.
+
+Extracted from webapp/query.py (donor_ledger.yaml: wa-query-search, wa-query-stats).
+Read-only: writes belong to AnalysisService.
+"""
+
+from __future__ import annotations
+
+import json
+import re
+from collections import Counter
+from pathlib import Path
+
+from sqlalchemy import and_, func, or_, select
+from sqlalchemy.orm import sessionmaker
+
+from photo_pipeline.models import AnalysisResult, Asset
+
+CARD_FIELDS = (
+    "status",
+    "description",
+    "people_count",
+    "setting",
+    "time_of_day",
+    "season",
+    "mood",
+    "location_hint",
+    "approx_year",
+)
+DONE = ("analyzed",)
+_SORTS = {
+    "year": (AnalysisResult.approx_year.desc(), Asset.current_path),
+    "people": (AnalysisResult.people_count.desc(), Asset.current_path),
+    "recent": (AnalysisResult.analyzed_at.desc(), Asset.current_path),
+    "path": (Asset.current_path,),
+}
+
+
+def _album_of(path: str | None) -> str:
+    return Path(path).parent.name if path else "(unknown)"
+
+
+class LibraryService:
+    def __init__(self, session_factory: sessionmaker) -> None:
+        self._session_factory = session_factory
+
+    def search(self, q="", filters=None, sort="path", offset=0, limit=60) -> dict:
+        filters = filters or {}
+        with self._session_factory() as session:
+            stmt = select(AnalysisResult, Asset.current_path).join(
+                Asset, Asset.id == AnalysisResult.asset_id
+            )
+            conds = _filter_conditions(filters)
+            for token in re.findall(r"\w+", q, re.UNICODE):
+                like = f"%{token}%"
+                conds.append(
+                    or_(AnalysisResult.description.ilike(like), AnalysisResult.tags.ilike(like))
+                )
+            if conds:
+                stmt = stmt.where(and_(*conds))
+            total = session.scalar(select(func.count()).select_from(stmt.subquery()))
+            stmt = stmt.order_by(*_SORTS.get(sort, _SORTS["path"])).limit(limit).offset(offset)
+            rows = [_card(result, path) for result, path in session.execute(stmt)]
+        return {"rows": rows, "total": total, "offset": offset, "limit": limit}
+
+    def stats(self) -> dict:
+        with self._session_factory() as session:
+            status = dict(
+                session.execute(
+                    select(AnalysisResult.status, func.count()).group_by(AnalysisResult.status)
+                ).all()
+            )
+            rows = list(
+                session.execute(
+                    select(AnalysisResult, Asset.current_path).join(
+                        Asset, Asset.id == AnalysisResult.asset_id
+                    )
+                )
+            )
+            albums: dict[str, dict] = {}
+            tag_counts: Counter = Counter()
+            year_counts: Counter = Counter()
+            people: Counter = Counter()
+            errors = []
+            for result, path in rows:
+                album = _album_of(path)
+                bucket = albums.setdefault(album, {"album": album, "done": 0, "total": 0})
+                bucket["total"] += 1
+                if result.status in DONE:
+                    bucket["done"] += 1
+                for tag in _tags(result.tags):
+                    tag_counts[tag] += 1
+                if result.approx_year is not None:
+                    year_counts[result.approx_year] += 1
+                if result.people_count is not None:
+                    people["3+" if result.people_count >= 3 else str(result.people_count)] += 1
+                if result.status == "error":
+                    errors.append({"path": path, "error": result.error_message})
+        return {
+            "total": sum(status.values()),
+            "status": status,
+            "setting": self._facet("setting"),
+            "time_of_day": self._facet("time_of_day"),
+            "season": self._facet("season"),
+            "people": [{"value": v, "count": n} for v, n in sorted(people.items())],
+            "years": [{"value": y, "count": year_counts[y]} for y in sorted(year_counts)],
+            "top_tags": [{"value": t, "count": n} for t, n in tag_counts.most_common(40)],
+            "albums": sorted(albums.values(), key=lambda d: d["album"]),
+            "errors": sorted(errors, key=lambda e: e["path"] or ""),
+        }
+
+    def facets(self) -> dict:
+        return {
+            "setting": self._facet("setting"),
+            "time_of_day": self._facet("time_of_day"),
+            "season": self._facet("season"),
+            "status": self._facet("status"),
+        }
+
+    def _facet(self, field: str) -> list[dict]:
+        column = getattr(AnalysisResult, field)
+        with self._session_factory() as session:
+            rows = session.execute(
+                select(column, func.count())
+                .where(column.is_not(None), func.trim(column) != "")
+                .group_by(column)
+                .order_by(func.count().desc())
+            ).all()
+        return [{"value": value, "count": count} for value, count in rows]
+
+
+def _filter_conditions(filters: dict) -> list:
+    conds = []
+    for key, column in (
+        ("setting", AnalysisResult.setting),
+        ("tod", AnalysisResult.time_of_day),
+        ("season", AnalysisResult.season),
+        ("status", AnalysisResult.status),
+    ):
+        if filters.get(key):
+            conds.append(column == filters[key])
+    people = filters.get("people")
+    if people == "3+":
+        conds.append(AnalysisResult.people_count >= 3)
+    elif people in ("0", "1", "2"):
+        conds.append(AnalysisResult.people_count == int(people))
+    if filters.get("year_min"):
+        conds.append(AnalysisResult.approx_year >= int(filters["year_min"]))
+    if filters.get("year_max"):
+        conds.append(AnalysisResult.approx_year <= int(filters["year_max"]))
+    if filters.get("has_location"):
+        conds.append(
+            and_(
+                AnalysisResult.location_hint.is_not(None),
+                func.trim(AnalysisResult.location_hint) != "",
+                func.lower(AnalysisResult.location_hint) != "null",
+            )
+        )
+    return conds
+
+
+def _card(result: AnalysisResult, path: str | None) -> dict:
+    card = {field: getattr(result, field) for field in CARD_FIELDS}
+    card.update(
+        asset_id=result.asset_id,
+        current_path=path,
+        album=_album_of(path),
+        tags=_tags(result.tags),
+    )
+    return card
+
+
+def _tags(raw: str | None) -> list[str]:
+    if not raw:
+        return []
+    try:
+        return json.loads(raw)
+    except (ValueError, TypeError):
+        return []
--- a/photo_pipeline/services/safety.py
+++ b/photo_pipeline/services/safety.py
@@ -98,3 +98,202 @@ def exif_projection(decision: str) -> dict[str, list[str]]:
    if decision == SFW:
        return {"add": [SFW], "remove": [NSFW]}
    return {"add": [], "remove": []}
+
+
+# ── Persistence, scoring, and the durable safety decision ────────────────────
+
+import uuid
+from datetime import datetime, timezone
+
+from sqlalchemy import select
+from sqlalchemy.orm import sessionmaker
+
+from photo_pipeline.integrations import exiftool
+from photo_pipeline.models import Asset, SafetyReview
+from photo_pipeline.services import hashing
+
+DECISIONS = {SFW, NSFW, "deferred"}
+
+
+class SafetyError(Exception):
+    """Invalid safety operation (bad decision, unknown asset)."""
+
+
+def _now() -> datetime:
+    # Microsecond precision so the latest-review-wins ordering never ties, even
+    # when a decision is revised twice in the same DB-second (SQLite CURRENT_TIMESTAMP
+    # is only second-granular).
+    return datetime.now(timezone.utc)
+
+
+class SafetyService:
+    """Durable safety scoring, review decisions, and the EXIF safety checkpoint.
+
+    Scores and decisions live in ``safety_reviews`` keyed by ``asset_id`` (the
+    donor's path-keyed CSV is gone). The latest row per asset is the current
+    decision; ``AnalysisService`` reads it as the privacy gate.
+    """
+
+    def __init__(self, session_factory: sessionmaker, *, model=None) -> None:
+        self._session_factory = session_factory
+        self._model = model  # injected in tests; real NsfwModel is lazy-loaded
+
+    # -- reads ----------------------------------------------------------------
+    def _latest_by_asset(self, session) -> dict[str, SafetyReview]:
+        # Latest row per asset. Small local scale: order ascending, let later rows
+        # overwrite. ponytail: a windowed query if safety_reviews ever grows huge.
+        latest: dict[str, SafetyReview] = {}
+        for review in session.scalars(select(SafetyReview).order_by(SafetyReview.created_at)):
+            latest[review.asset_id] = review
+        return latest
+
+    def current_decision(self, asset_id: str) -> str | None:
+        with self._session_factory() as session:
+            review = self._latest_by_asset(session).get(asset_id)
+            return review.decision if review else None
+
+    def counts(self) -> dict[str, int]:
+        """Decision breakdown over canonical, active assets — the workflow totals."""
+        with self._session_factory() as session:
+            assets = list(session.scalars(_eligible_assets_query()))
+            latest = self._latest_by_asset(session)
+            out = {SFW: 0, NSFW: 0, "deferred": 0, "undecided": 0, "scored": 0}
+            for asset in assets:
+                review = latest.get(asset.id)
+                decision = review.decision if review else None
+                if decision in (SFW, NSFW, "deferred"):
+                    out[decision] += 1
+                else:
+                    out["undecided"] += 1
+                    if review and review.score is not None:
+                        out["scored"] += 1
+            return out
+
+    def review_queue(self, state: str = "", limit: int = 100, offset: int = 0) -> dict:
+        """Assets for the review UI, filtered by ``state`` (undecided/sfw/nsfw/deferred)."""
+        with self._session_factory() as session:
+            assets = list(session.scalars(_eligible_assets_query().order_by(Asset.current_path)))
+            latest = self._latest_by_asset(session)
+            rows = []
+            for asset in assets:
+                review = latest.get(asset.id)
+                decision = review.decision if review else None
+                effective = decision or "undecided"
+                if state and state != effective:
+                    continue
+                rows.append(
+                    {
+                        "asset_id": asset.id,
+                        "current_path": asset.current_path,
+                        "score": review.score if review else None,
+                        "decision": decision,
+                        "suggested": classify(review.score) if review and review.score is not None else None,
+                        "exif_verified": bool(review and review.exif_verified_at),
+                    }
+                )
+            return {"total": len(rows), "items": rows[offset : offset + limit]}
+
+    def scorable_asset_ids(self) -> list[str]:
+        """Canonical active assets with a path — the items a scoring job enqueues."""
+        with self._session_factory() as session:
+            return [a.id for a in session.scalars(_eligible_assets_query()) if a.current_path]
+
+    # -- writes ---------------------------------------------------------------
+    def score_assets(self, asset_ids: list[str] | None = None) -> int:
+        """Score eligible assets with the local model and persist scored reviews.
+
+        Records score-only reviews (no decision) so the reviewer can sort by risk.
+        Returns the number scored. Assets already carrying a human decision keep it.
+        """
+        with self._session_factory() as session:
+            query = _eligible_assets_query()
+            if asset_ids is not None:
+                query = query.where(Asset.id.in_(asset_ids))
+            assets = [a for a in session.scalars(query) if a.current_path]
+        if not assets:
+            return 0
+        model = self._model or _default_model()
+        by_path = {a.current_path: a.id for a in assets}
+        scored = model.score(list(by_path))
+        now = _now()
+        with self._session_factory() as session:
+            for path, score in scored:
+                session.add(
+                    SafetyReview(
+                        id=str(uuid.uuid4()), asset_id=by_path[path], score=float(score), created_at=now
+                    )
+                )
+            session.commit()
+        return len(scored)
+
+    def decide(self, asset_id: str, decision: str, *, reviewer: str = "user", write_exif: bool = True) -> dict:
+        """Persist a human safety decision and, for sfw/nsfw, run the EXIF checkpoint.
+
+        The decision is durable regardless of EXIF; ``exif_verified`` becomes true
+        only after the mutually-exclusive keyword is written and read back, and the
+        file's new SHA-256 is stored — that verified state is what upload eligibility
+        later depends on.
+        """
+        if decision not in DECISIONS:
+            raise SafetyError(f"invalid decision {decision!r}")
+        with self._session_factory() as session:
+            asset = session.get(Asset, asset_id)
+            if asset is None:
+                raise SafetyError(f"unknown asset {asset_id}")
+            prior = self._latest_by_asset(session).get(asset_id)
+            path = asset.current_path
+
+        exif_verified_at = None
+        result_sha256 = None
+        if write_exif and decision in (SFW, NSFW) and path:
+            ops = exif_projection(decision)
+            if exiftool.apply_keywords(path, add=ops["add"], remove=ops["remove"]):
+                # Read back: the chosen keyword present, the opposite absent.
+                keywords = exiftool.read_keyword_sets([path]).get(path, set())
+                opposite = NSFW if decision == SFW else SFW
+                if decision in keywords and opposite not in keywords:
+                    exif_verified_at = _now()
+                    result_sha256 = hashing.sha256_file(path)
+
+        now = _now()
+        with self._session_factory() as session:
+            review = SafetyReview(
+                id=str(uuid.uuid4()),
+                asset_id=asset_id,
+                decision=decision,
+                prior_decision=prior.decision if prior else None,
+                reviewer=reviewer,
+                score=prior.score if prior else None,
+                exif_verified_at=exif_verified_at,
+                result_sha256=result_sha256,
+                created_at=now,
+            )
+            session.add(review)
+            if result_sha256:
+                asset = session.get(Asset, asset_id)
+                asset.current_sha256 = result_sha256
+            session.commit()
+        return {
+            "asset_id": asset_id,
+            "decision": decision,
+            "prior_decision": prior.decision if prior else None,
+            "exif_verified": exif_verified_at is not None,
+        }
+
+
+def _eligible_assets_query():
+    """Canonical, active assets — the safety stage runs only on these.
+
+    ``canonical_asset_id IS NULL`` excludes non-canonical duplicate variants; the
+    shared discovery service already excludes ``_IGNORE/``.
+    """
+    return select(Asset).where(
+        Asset.canonical_asset_id.is_(None),
+        Asset.availability_state == "active",
+    )
+
+
+def _default_model():
+    from photo_pipeline.integrations.nsfw_model import NsfwModel
+
+    return NsfwModel()
--- a/photo_pipeline/services/workflow.py
+++ b/photo_pipeline/services/workflow.py
@@ -0,0 +1,161 @@
+"""WorkflowService — stage readiness for the Workflow home.
+
+Totals are DERIVED from the source tables (assets, duplicate_clusters,
+safety_reviews, analysis_results) rather than a separately-maintained counter, so
+the home page can never drift from reality. Each stage card carries a machine
+``status`` plus human ``status_label`` and ``blocker`` text — the UI communicates
+state with text/icons as well as color (concept §Workflow home).
+
+``active_job`` is the one-mutating-job indicator: when a job holds the
+``library_write`` lock, mutating actions across the app are disabled with a reason,
+while read-only browsing stays available.
+"""
+
+from __future__ import annotations
+
+from sqlalchemy import func, select
+from sqlalchemy.orm import sessionmaker
+
+from photo_pipeline.jobs.domain_handlers import ANALYSIS, SAFETY_SCORE
+from photo_pipeline.models import Asset, DuplicateCluster, Job
+from photo_pipeline.services.analysis import AnalysisService
+from photo_pipeline.services.jobs import ACTIVE_STATES
+from photo_pipeline.services.safety import SafetyService
+
+# Job type that produced each stage's last run.
+STAGE_JOB_TYPE = {"inventory": "scan", "safety": SAFETY_SCORE, "analysis": ANALYSIS}
+
+
+class WorkflowService:
+    def __init__(self, session_factory: sessionmaker) -> None:
+        self._session_factory = session_factory
+
+    def readiness(self) -> dict:
+        with self._session_factory() as session:
+            asset_total = session.scalar(
+                select(func.count()).select_from(Asset).where(Asset.canonical_asset_id.is_(None))
+            )
+            missing = session.scalar(
+                select(func.count()).select_from(Asset).where(Asset.missing_at.is_not(None))
+            )
+            cluster_states = dict(
+                session.execute(
+                    select(DuplicateCluster.state, func.count()).group_by(DuplicateCluster.state)
+                ).all()
+            )
+            last_run = self._last_runs(session)
+            active = self._active_job(session)
+
+        safety = SafetyService(self._session_factory).counts()
+        analysis = AnalysisService(self._session_factory).counts()
+        undecided_clusters = cluster_states.get("open", 0) + cluster_states.get("reopened", 0)
+
+        stages = [
+            _card(
+                "inventory",
+                "Inventory",
+                status="complete" if asset_total and not missing else ("attention" if missing else "ready"),
+                counts={"assets": asset_total or 0, "missing": missing or 0},
+                blocker=f"{missing} missing file(s)" if missing else None,
+                last_run=last_run.get("inventory"),
+                action={"label": "Rescan", "job_type": "scan", "route": "#/inventory"},
+            ),
+            _card(
+                "duplicates",
+                "Duplicates",
+                status="attention" if undecided_clusters else "complete",
+                counts={"undecided": undecided_clusters, "total": sum(cluster_states.values())},
+                blocker=f"{undecided_clusters} cluster(s) to review" if undecided_clusters else None,
+                last_run=None,
+                action={"label": "Review", "route": "#/duplicates"},
+            ),
+            _card(
+                "safety",
+                "Safety",
+                # Blocked while duplicates are unreviewed (dedup precedes safety).
+                status=_safety_status(safety, undecided_clusters),
+                counts=safety,
+                blocker=(
+                    "resolve duplicate clusters first"
+                    if undecided_clusters
+                    else (f"{safety['undecided']} undecided" if safety["undecided"] else None)
+                ),
+                last_run=last_run.get("safety"),
+                action={"label": "Review safety", "job_type": SAFETY_SCORE, "route": "#/safety"},
+                blocked_by="duplicates" if undecided_clusters else None,
+            ),
+            _card(
+                "analysis",
+                "Analysis",
+                status=_analysis_status(analysis, safety),
+                counts=analysis,
+                blocker=(
+                    "no confirmed-SFW assets yet"
+                    if analysis["eligible"] == 0
+                    else (f"{analysis['pending']} pending" if analysis["pending"] else None)
+                ),
+                last_run=last_run.get("analysis"),
+                action={"label": "Analyze", "job_type": ANALYSIS, "route": "#/analyze"},
+                blocked_by="safety" if analysis["eligible"] == 0 else None,
+            ),
+        ]
+        return {"active_job": active, "stages": stages}
+
+    def _last_runs(self, session) -> dict[str, str | None]:
+        out: dict[str, str | None] = {}
+        for stage, job_type in STAGE_JOB_TYPE.items():
+            finished = session.scalar(
+                select(func.max(Job.finished_at)).where(Job.job_type == job_type)
+            )
+            out[stage] = finished.isoformat() if finished else None
+        return out
+
+    def _active_job(self, session) -> dict | None:
+        job = session.scalars(
+            select(Job).where(Job.state.in_(ACTIVE_STATES)).order_by(Job.created_at.desc())
+        ).first()
+        if job is None:
+            return None
+        return {"id": job.id, "job_type": job.job_type, "state": job.state, "lock": job.lock_key}
+
+
+def _safety_status(safety: dict, undecided_clusters: int) -> str:
+    if undecided_clusters:
+        return "blocked"
+    if safety["undecided"]:
+        return "attention"
+    if safety[_SFW] or safety[_NSFW]:
+        return "complete"
+    return "ready"
+
+
+def _analysis_status(analysis: dict, safety: dict) -> str:
+    if analysis["eligible"] == 0:
+        return "blocked"
+    if analysis["pending"]:
+        return "attention"
+    return "complete"
+
+
+def _card(key, title, *, status, counts, blocker, last_run, action, blocked_by=None) -> dict:
+    return {
+        "key": key,
+        "title": title,
+        "status": status,
+        "status_label": _STATUS_LABEL[status],
+        "counts": counts,
+        "blocker": blocker,
+        "blocked_by": blocked_by,
+        "last_run": last_run,
+        "action": action,
+    }
+
+
+_SFW = "sfw"
+_NSFW = "nsfw"
+_STATUS_LABEL = {
+    "complete": "Complete",
+    "ready": "Ready",
+    "attention": "Needs review",
+    "blocked": "Blocked",
+}
--- a/tests/e2e/test_workflow_views.py
+++ b/tests/e2e/test_workflow_views.py
@@ -0,0 +1,162 @@
+"""Browser journeys for the Phase B views (US02-06): Workflow home, Safety review,
+Library, Analyze, Stats — plus the read-only-during-jobs rule.
+
+The library is seeded in-process (scan → one SFW/one NSFW decision → one fake-analyzed
+result) before the real server starts, so the views have data to render exactly as
+they would in production.
+"""
+
+import socket
+import subprocess
+import sys
+import time
+from pathlib import Path
+from types import SimpleNamespace
+
+import httpx
+import numpy as np
+import pytest
+from PIL import Image
+
+REPO = Path(__file__).resolve().parents[2]
+
+
+def _image(path, seed):
+    rng = np.random.default_rng(seed)
+    Image.fromarray(rng.integers(0, 256, (96, 128, 3), dtype=np.uint8)).save(path, quality=90)
+
+
+def _free_port():
+    with socket.socket() as sock:
+        sock.bind(("127.0.0.1", 0))
+        return sock.getsockname()[1]
+
+
+class _Fake:
+    def analyze(self, path, *, album_hint):
+        return {"description": "a sunny beach at golden hour", "tags": ["beach", "sunset"], "people_count": 2, "setting": "outdoor"}
+
+
+@pytest.fixture
+def server(tmp_path):
+    data = tmp_path / "data"
+    data.mkdir()
+    lib = tmp_path / "lib"
+    lib.mkdir()
+    _image(lib / "beach.jpg", 1)
+    _image(lib / "city.jpg", 2)
+    _image(lib / "night.jpg", 3)
+
+    from photo_pipeline.config import Config
+    from photo_pipeline.db import create_db_engine, create_session_factory, run_migrations
+    from photo_pipeline.services.analysis import AnalysisService
+    from photo_pipeline.services.inventory import InventoryService
+    from photo_pipeline.services.safety import SafetyService
+
+    config = Config.from_env(
+        {"PHOTO_PIPELINE_DATA_DIR": str(data), "PHOTO_PIPELINE_LIBRARY_ROOTS": str(lib)}
+    )
+    run_migrations(config.database_url)
+    engine = create_db_engine(config.database_url)
+    sf = create_session_factory(engine)
+    InventoryService(sf).scan(lib)
+    from sqlalchemy import select
+
+    from photo_pipeline.models import Asset
+
+    with sf() as session:
+        ids = list(session.scalars(select(Asset.id).order_by(Asset.current_path)))
+    safety = SafetyService(sf)
+    safety.decide(ids[0], "sfw", write_exif=False)
+    safety.decide(ids[1], "nsfw", write_exif=False)
+    AnalysisService(sf, provider=_Fake()).run([ids[0]])
+    engine.dispose()
+
+    port = _free_port()
+    env = {
+        "PATH": __import__("os").environ.get("PATH", ""),
+        "PHOTO_PIPELINE_DATA_DIR": str(data),
+        "PHOTO_PIPELINE_LIBRARY_ROOTS": str(lib),
+        "PHOTO_PIPELINE_HOST": "127.0.0.1",
+        "PHOTO_PIPELINE_PORT": str(port),
+    }
+    proc = subprocess.Popen(
+        [sys.executable, "-m", "photo_pipeline", "serve"],
+        cwd=str(REPO),
+        env=env,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+    )
+    base = f"http://127.0.0.1:{port}"
+    deadline = time.monotonic() + 30
+    ready = False
+    while time.monotonic() < deadline:
+        if proc.poll() is not None:
+            out, err = proc.communicate()
+            pytest.fail(f"server exited: {err.decode(errors='replace')}")
+        try:
+            if httpx.get(f"{base}/api/v1/health/ready", timeout=1).status_code == 200:
+                ready = True
+                break
+        except httpx.HTTPError:
+            time.sleep(0.2)
+    if not ready:
+        proc.terminate()
+        pytest.fail("server never became ready")
+    try:
+        yield SimpleNamespace(base=base)
+    finally:
+        proc.terminate()
+        try:
+            proc.wait(timeout=10)
+        except subprocess.TimeoutExpired:
+            proc.kill()
+
+
+def test_workflow_home_shows_cards_with_counts_and_labels(page, server):
+    page.goto(f"{server.base}/app/#/workflow")
+    page.get_by_test_id("stage-safety").wait_for()
+    # Status is text, not color alone.
+    assert page.get_by_test_id("status-safety").inner_text().strip() != ""
+    # 1 sfw, 1 nsfw, 1 undecided from the seed.
+    assert "1 undecided" in page.get_by_test_id("counts-safety").inner_text()
+    assert page.get_by_test_id("stage-analysis").count() == 1
+    assert page.get_by_test_id("action-safety").is_visible()
+
+
+def test_actions_disabled_and_explained_while_a_job_runs(page, server):
+    # Start a durable job (no worker consumes it), then the home page must disable
+    # job-starting actions and explain why.
+    httpx.post(f"{server.base}/api/v1/safety/jobs", timeout=5).raise_for_status()
+    page.goto(f"{server.base}/app/#/workflow")
+    page.get_by_test_id("job-running").wait_for()
+    assert page.get_by_test_id("action-safety").is_disabled()
+
+
+def test_safety_review_decide_persists_across_reload(page, server):
+    page.goto(f"{server.base}/app/#/safety?state=undecided")
+    row = page.get_by_test_id("safety-row").first
+    row.wait_for()
+    asset_id = row.get_attribute("data-asset-id")
+    row.get_by_test_id("decide-sfw").click()
+
+    # The decided asset leaves the undecided queue and appears under SFW after reload.
+    page.goto(f"{server.base}/app/#/safety?state=sfw")
+    page.get_by_test_id("safety-body").wait_for()
+    page.wait_for_selector(f'[data-asset-id="{asset_id}"]')
+
+
+def test_library_search_and_stats(page, server):
+    page.goto(f"{server.base}/app/#/library?q=beach")
+    page.get_by_test_id("library-card").first.wait_for()
+    assert "beach" in page.get_by_test_id("library-card").first.inner_text().lower()
+
+    page.goto(f"{server.base}/app/#/stats")
+    page.get_by_test_id("stats-albums").wait_for()
+    assert page.get_by_test_id("stats-status").inner_text().strip() != ""
+
+
+def test_analyze_view_shows_counts(page, server):
+    page.goto(f"{server.base}/app/#/analyze")
+    page.get_by_test_id("analyze-counts").wait_for()
+    assert page.get_by_test_id("run-analysis").is_visible()
--- a/tests/integration/test_safety_analysis.py
+++ b/tests/integration/test_safety_analysis.py
@@ -0,0 +1,146 @@
+"""Safety decisions, the NSFW→analysis privacy gate, and one-mutating-job policy
+(US02-06).
+
+The privacy invariant is the headline: the vision provider is called ONLY for
+confirmed-SFW assets. A call-recording fake provider proves nsfw/undecided assets
+never produce a request, with no network or API key.
+"""
+
+import shutil
+import uuid
+from datetime import datetime, timezone
+
+import pytest
+from fastapi.testclient import TestClient
+from PIL import Image
+
+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
+from photo_pipeline.services.analysis import AnalysisService
+from photo_pipeline.services.safety import SafetyService
+
+
+def _factory(tmp_path):
+    (tmp_path / "data").mkdir()
+    config = Config.from_env({"PHOTO_PIPELINE_DATA_DIR": str(tmp_path / "data")})
+    run_migrations(config.database_url)
+    return config, create_session_factory(create_db_engine(config.database_url))
+
+
+def _seed_assets(sf, paths):
+    now = datetime.now(timezone.utc)
+    ids = []
+    with sf() as session:
+        for path in paths:
+            asset = Asset(
+                id=str(uuid.uuid4()),
+                original_path=str(path),
+                current_path=str(path),
+                discovered_at=now,
+                hash_version=1,
+            )
+            session.add(asset)
+            ids.append(asset.id)
+        session.commit()
+    return ids
+
+
+class RecordingProvider:
+    def __init__(self):
+        self.calls = []
+
+    def analyze(self, path, *, album_hint):
+        self.calls.append(path)
+        return {"description": "a photo", "tags": ["alpha", "beta"], "people_count": 1}
+
+
+def test_provider_called_only_for_confirmed_sfw(tmp_path):
+    _, sf = _factory(tmp_path)
+    sfw, nsfw, undecided = _seed_assets(sf, ["/lib/a.jpg", "/lib/b.jpg", "/lib/c.jpg"])
+    safety = SafetyService(sf)
+    safety.decide(sfw, "sfw", write_exif=False)
+    safety.decide(nsfw, "nsfw", write_exif=False)
+    # `undecided` gets no decision at all.
+
+    provider = RecordingProvider()
+    result = AnalysisService(sf, provider=provider).run([sfw, nsfw, undecided])
+
+    assert provider.calls == ["/lib/a.jpg"], "provider must see only the SFW asset"
+    assert result == {"analyzed": 1, "skipped": 2, "errors": 0}
+
+
+def test_flipping_sfw_to_nsfw_removes_analysis_eligibility(tmp_path):
+    _, sf = _factory(tmp_path)
+    (asset,) = _seed_assets(sf, ["/lib/a.jpg"])
+    safety = SafetyService(sf)
+    analysis = AnalysisService(sf, provider=RecordingProvider())
+
+    safety.decide(asset, "sfw", write_exif=False)
+    assert analysis.eligible_asset_ids() == [asset]
+    safety.decide(asset, "nsfw", write_exif=False)
+    assert analysis.eligible_asset_ids() == []  # latest decision wins
+
+
+def test_decision_persists_and_shows_in_queue(tmp_path):
+    config, sf = _factory(tmp_path)
+    (asset,) = _seed_assets(sf, ["/lib/a.jpg"])
+    app = create_app(config)
+    with TestClient(app) as client:
+        ok = client.post("/api/v1/safety/decisions", json={"asset_id": asset, "decision": "nsfw"})
+        assert ok.status_code == 200 and ok.json()["decision"] == "nsfw"
+
+        nsfw_queue = client.get("/api/v1/safety/queue", params={"state": "nsfw"}).json()
+        assert [row["asset_id"] for row in nsfw_queue["items"]] == [asset]
+        assert client.get("/api/v1/safety/counts").json()["nsfw"] == 1
+
+    unknown = TestClient(app)
+    with unknown as client:
+        bad = client.post("/api/v1/safety/decisions", json={"asset_id": "nope", "decision": "sfw"})
+        assert bad.status_code == 404
+
+
+def test_one_mutating_job_at_a_time(tmp_path):
+    config, sf = _factory(tmp_path)
+    ids = _seed_assets(sf, ["/lib/a.jpg", "/lib/b.jpg"])
+    SafetyService(sf).decide(ids[0], "sfw", write_exif=False)  # make analysis eligible
+    app = create_app(config)
+    with TestClient(app) as client:
+        first = client.post("/api/v1/safety/jobs")
+        assert first.status_code == 200
+        # Both scoring and analysis take the library_write lock — the second is refused.
+        second = client.post("/api/v1/analysis/jobs")
+        assert second.status_code == 409
+
+        workflow = client.get("/api/v1/workflow").json()
+        assert workflow["active_job"] is not None
+        assert workflow["active_job"]["lock"] == "library_write"
+
+
+def test_workflow_counts_reflect_decisions(tmp_path):
+    config, sf = _factory(tmp_path)
+    ids = _seed_assets(sf, ["/lib/a.jpg", "/lib/b.jpg", "/lib/c.jpg"])
+    SafetyService(sf).decide(ids[0], "sfw", write_exif=False)
+    app = create_app(config)
+    with TestClient(app) as client:
+        stages = {s["key"]: s for s in client.get("/api/v1/workflow").json()["stages"]}
+        assert stages["safety"]["counts"] == {"sfw": 1, "nsfw": 0, "deferred": 0, "undecided": 2, "scored": 0}
+        assert stages["safety"]["status"] == "attention"
+        assert stages["analysis"]["counts"]["eligible"] == 1
+
+
+@pytest.mark.skipif(shutil.which("exiftool") is None, reason="exiftool not installed")
+def test_safety_decision_writes_and_verifies_exif(tmp_path):
+    _, sf = _factory(tmp_path)
+    image = tmp_path / "photo.jpg"
+    Image.new("RGB", (32, 32), (120, 60, 30)).save(image)
+    (asset,) = _seed_assets(sf, [image])
+
+    result = SafetyService(sf).decide(asset, "nsfw")
+    assert result["exif_verified"] is True
+
+    from photo_pipeline.integrations import exiftool
+
+    keywords = exiftool.read_keyword_sets([str(image)]).get(str(image), set())
+    assert "nsfw" in keywords and "sfw" not in keywords
--- a/tests/story_traceability.json
+++ b/tests/story_traceability.json
@@ -57,6 +57,13 @@
    "US02-04": [
      "tests/integration/test_jobs_api.py",
      "tests/integration/test_jobs_sse.py"
+    ],
+    "US02-05": [
+      "tests/e2e/test_frontend_shell.py"
+    ],
+    "US02-06": [
+      "tests/integration/test_safety_analysis.py",
+      "tests/e2e/test_workflow_views.py"
    ]
  }
 }