diff --git a/frontend/css/app.css b/frontend/css/app.css
index 6ad199b..e793e78 100644
--- 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;
+}
diff --git a/frontend/index.html b/frontend/index.html
index bfeb057..2f2f2a7 100644
--- a/frontend/index.html
+++ b/frontend/index.html
@@ -10,8 +10,13 @@
diff --git a/frontend/js/api.js b/frontend/js/api.js
index 4648764..e65f099 100644
--- 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),
};
diff --git a/frontend/js/app.js b/frontend/js/app.js
index e70ab62..f999c57 100644
--- 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();
diff --git a/frontend/js/dom.js b/frontend/js/dom.js
new file mode 100644
index 0000000..b7827ae
--- /dev/null
+++ 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");
+ });
+}
diff --git a/frontend/js/router.js b/frontend/js/router.js
index 939128c..5608eb7 100644
--- 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)) };
}
diff --git a/frontend/js/tests/unit.js b/frontend/js/tests/unit.js
index d24fdf7..1293797 100644
--- 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 ────────────────────────────────────────────
diff --git a/frontend/js/views.js b/frontend/js/views.js
new file mode 100644
index 0000000..a7de115
--- /dev/null
+++ 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;
+}
diff --git a/migrations/versions/0005_safety_analysis_workflow.py b/migrations/versions/0005_safety_analysis_workflow.py
new file mode 100644
index 0000000..b11abf4
--- /dev/null
+++ 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")
diff --git a/photo_pipeline/api/app.py b/photo_pipeline/api/app.py
index 9be1229..940ea14 100644
--- 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")
diff --git a/photo_pipeline/api/routes/analysis.py b/photo_pipeline/api/routes/analysis.py
new file mode 100644
index 0000000..03d12d7
--- /dev/null
+++ 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))
diff --git a/photo_pipeline/api/routes/library.py b/photo_pipeline/api/routes/library.py
new file mode 100644
index 0000000..5fd949c
--- /dev/null
+++ 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()
diff --git a/photo_pipeline/api/routes/safety.py b/photo_pipeline/api/routes/safety.py
new file mode 100644
index 0000000..f4a2fe2
--- /dev/null
+++ 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))
diff --git a/photo_pipeline/api/routes/workflow.py b/photo_pipeline/api/routes/workflow.py
new file mode 100644
index 0000000..d6a37cc
--- /dev/null
+++ 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()
diff --git a/photo_pipeline/jobs/domain_handlers.py b/photo_pipeline/jobs/domain_handlers.py
new file mode 100644
index 0000000..6aa5252
--- /dev/null
+++ 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)
diff --git a/photo_pipeline/jobs/handlers.py b/photo_pipeline/jobs/handlers.py
index 57f9509..3532e42 100644
--- 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
diff --git a/photo_pipeline/jobs/worker.py b/photo_pipeline/jobs/worker.py
index f7b660c..e70106e 100644
--- 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
diff --git a/photo_pipeline/models/__init__.py b/photo_pipeline/models/__init__.py
index 392a45e..58e746c 100644
--- 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",
]
diff --git a/photo_pipeline/models/workflow.py b/photo_pipeline/models/workflow.py
new file mode 100644
index 0000000..6eb87ad
--- /dev/null
+++ 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))
diff --git a/photo_pipeline/schemas/__init__.py b/photo_pipeline/schemas/__init__.py
index 0c0822d..10979f3 100644
--- 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"]
diff --git a/photo_pipeline/schemas/safety.py b/photo_pipeline/schemas/safety.py
new file mode 100644
index 0000000..6d124f7
--- /dev/null
+++ 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"]
diff --git a/photo_pipeline/services/analysis.py b/photo_pipeline/services/analysis.py
new file mode 100644
index 0000000..8222f2b
--- /dev/null
+++ 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"
diff --git a/photo_pipeline/services/library.py b/photo_pipeline/services/library.py
new file mode 100644
index 0000000..04a940a
--- /dev/null
+++ 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 []
diff --git a/photo_pipeline/services/safety.py b/photo_pipeline/services/safety.py
index 326bbbe..5983c45 100644
--- 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()
diff --git a/photo_pipeline/services/workflow.py b/photo_pipeline/services/workflow.py
new file mode 100644
index 0000000..3308945
--- /dev/null
+++ 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",
+}
diff --git a/tests/e2e/test_workflow_views.py b/tests/e2e/test_workflow_views.py
new file mode 100644
index 0000000..28f7e12
--- /dev/null
+++ 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()
diff --git a/tests/integration/test_safety_analysis.py b/tests/integration/test_safety_analysis.py
new file mode 100644
index 0000000..0f2b514
--- /dev/null
+++ 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
diff --git a/tests/story_traceability.json b/tests/story_traceability.json
index d405bf9..c0a3bf2 100644
--- 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"
]
}
}