README.md

@@ -46,6 +46,6 @@ work_item/scripts/python -m pytest tests/e2e tests/integration -q
   exclusion, move reconciliation, exact/fuzzy duplicate review, thumbnail
   orientation, canonical selection, browser reload, and a full process restart —
   asserting durable API and database state after the restart.
-- `tests/phase_a_traceability.json` maps every story (US01-01 … US01-07) to its
-  tests; `tests/e2e/test_traceability.py` fails if a story loses coverage or a test
+- `tests/story_traceability.json` maps every delivered story to its tests;
+  `tests/e2e/test_traceability.py` fails if a Phase A story loses coverage or a test
   file is left unexercised.

frontend/css/app.css

@@ -1,16 +1,4 @@
-:root {
-  --bg: #000;
-  --surface: #0e0e11;
-  --surface-2: #17171c;
-  --border: #26262e;
-  --text: #f4f4f6;
-  --muted: #9a9aa6;
-  --accent: #5b8cff;
-  --danger: #ff5b6e;
-  --ok: #3ddc97;
-  --warn: #ffcf5b;
-  --radius: 10px;
-}
+@import "./tokens.css";
 
 * { box-sizing: border-box; }
 

frontend/css/tokens.css

@@ -0,0 +1,20 @@
+/* Shared design tokens — the reusable visual language extracted from the donor
+ * web UIs (nsfwtag review + Photo Analyzer webapp): the dark OLED palette, status
+ * colors, radius, and surfaces. Every view (inventory, duplicates, and the
+ * upcoming safety/analyze views) consumes these instead of hard-coded values, so
+ * the look stays consistent as the workflow shell grows.
+ *
+ * donor_ledger.yaml: nt-ui, wa-server, pa-ui-terminal (design tokens). */
+:root {
+  --bg: #000;
+  --surface: #0e0e11;
+  --surface-2: #17171c;
+  --border: #26262e;
+  --text: #f4f4f6;
+  --muted: #9a9aa6;
+  --accent: #5b8cff;
+  --danger: #ff5b6e;
+  --ok: #3ddc97;
+  --warn: #ffcf5b;
+  --radius: 10px;
+}

photo_pipeline/integrations/__init__.py

@@ -0,0 +1,5 @@
+"""External-tool adapters (exiftool, vision, NSFW model, immich-go).
+
+Integrations own all subprocess and network I/O; services depend on these
+interfaces, never on the archived CLI entry points.
+"""

photo_pipeline/integrations/exiftool.py

@@ -0,0 +1,70 @@
+"""exiftool adapter: read and apply EXIF keywords.
+
+Only the ``Keywords`` and ``Subject`` fields are read or written, so every other
+tag (caption, dates, GPS, camera, ratings) is preserved. Reading is batched over
+stdin; writing uses the idempotent ``-=``/``+=`` pattern so re-running never
+duplicates a keyword.
+
+Extracted from nsfwtag.exif (_read_keywords / write_keyword / remove_keyword) and
+photo_analyzer's batched keyword read (donor_ledger.yaml: nt-exif-marks,
+nt-apply-list, pa-nsfw-filter). No dependency on the archived entry points.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import subprocess
+from collections.abc import Iterable
+
+
+def read_keyword_sets(paths: Iterable[str]) -> dict[str, set[str]]:
+    """Map each path to its lowercased set of ``Keywords`` + ``Subject`` values.
+
+    One batched exiftool call (paths fed via stdin). Returns ``{}`` if exiftool is
+    unavailable — callers decide how to fail (analysis fails open, safety fails safe).
+    """
+    paths = list(paths)
+    if not paths:
+        return {}
+    want = {os.path.normpath(p): p for p in paths}
+    try:
+        result = subprocess.run(
+            ["exiftool", "-m", "-j", "-Keywords", "-Subject", "-@", "-"],
+            input="\n".join(paths),
+            capture_output=True,
+            text=True,
+        )
+    except FileNotFoundError:
+        return {}
+    out: dict[str, set[str]] = {}
+    try:
+        records = json.loads(result.stdout or "[]")
+    except ValueError:
+        return {}
+    for record in records:
+        values: list = []
+        for field in ("Keywords", "Subject"):
+            value = record.get(field)
+            if isinstance(value, list):
+                values += value
+            elif value is not None:
+                values.append(value)
+        key = want.get(os.path.normpath(record.get("SourceFile", "")))
+        if key is not None:
+            out[key] = {str(v).strip().lower() for v in values}
+    return out
+
+
+def apply_keywords(path: str, *, add: Iterable[str] = (), remove: Iterable[str] = ()) -> bool:
+    """Idempotently add/remove keywords in Keywords + Subject; preserve all else."""
+    args = ["exiftool", "-m", "-overwrite_original"]
+    for kw in remove:
+        args += [f"-Keywords-={kw}", f"-Subject-={kw}"]
+    for kw in add:
+        # remove-then-add makes the add idempotent (no duplicate on re-run).
+        args += [f"-Keywords-={kw}", f"-Keywords+={kw}", f"-Subject-={kw}", f"-Subject+={kw}"]
+    if len(args) == 3:
+        return True
+    args.append(path)
+    return subprocess.run(args, capture_output=True, text=True).returncode == 0

photo_pipeline/integrations/nsfw_model.py

@@ -0,0 +1,84 @@
+"""Local NSFW model adapter.
+
+Wraps the on-device ViT classifier behind a small interface so the safety service
+depends on ``NsfwModel.score`` rather than on torch/transformers or the archived
+CLI. Heavy imports are lazy: importing this module never loads the model, so the
+rest of the app (and its tests) stay light. Scoring itself is non-deterministic
+across hardware and is exercised in dedicated model tests, not the unit suite.
+
+Extracted from nsfwtag.scoring.score_images (donor_ledger.yaml: nt-score-model).
+The donor's CSV cache is dropped here — scores are persisted in the database by
+asset ID (safety repository), not a path-keyed CSV.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+MODEL_ID = "AdamCodd/vit-base-nsfw-detector"
+BATCH = 16
+
+
+class NsfwModel:
+    def __init__(self, model_id: str = MODEL_ID, batch: int = BATCH) -> None:
+        self.model_id = model_id
+        self.batch = batch
+        self._model = None
+        self._device = None
+        self._size = None
+        self._nsfw_idx = None
+
+    def _ensure_loaded(self) -> None:
+        if self._model is not None:
+            return
+        import torch
+        from transformers import AutoModelForImageClassification
+
+        self._device = "mps" if torch.backends.mps.is_available() else "cpu"
+        try:
+            model = AutoModelForImageClassification.from_pretrained(
+                self.model_id, local_files_only=True
+            )
+        except Exception:
+            model = AutoModelForImageClassification.from_pretrained(self.model_id)
+        self._model = model.to(self._device).eval()
+        self._nsfw_idx = next(
+            i for i, label in model.config.id2label.items() if label.lower() == "nsfw"
+        )
+        self._size = getattr(model.config, "image_size", 224)
+
+    def score(self, paths: list[Path | str]) -> list[tuple[str, float]]:
+        """Return ``[(path, nsfw_probability)]`` for each readable image."""
+        if not paths:
+            return []
+        self._ensure_loaded()
+        import numpy as np
+        import torch
+        from PIL import Image, ImageFile
+
+        ImageFile.LOAD_TRUNCATED_IMAGES = True
+
+        def preprocess(image):
+            image = image.convert("RGB").resize((self._size, self._size), Image.BILINEAR)
+            array = (np.asarray(image, dtype="float32") / 255.0 - 0.5) / 0.5
+            return torch.from_numpy(array).permute(2, 0, 1)
+
+        results: list[tuple[str, float]] = []
+        items = [str(p) for p in paths]
+        for start in range(0, len(items), self.batch):
+            tensors, batch_paths = [], []
+            for path in items[start : start + self.batch]:
+                try:
+                    tensors.append(preprocess(Image.open(path)))
+                    batch_paths.append(path)
+                except Exception:
+                    continue
+            if not tensors:
+                continue
+            with torch.no_grad():
+                probs = self._model(
+                    pixel_values=torch.stack(tensors).to(self._device)
+                ).logits.softmax(-1)
+            for path, prob in zip(batch_paths, probs):
+                results.append((path, float(prob[self._nsfw_idx].item())))
+        return results

photo_pipeline/services/safety.py

@@ -0,0 +1,100 @@
+"""Safety (NSFW) logic: scoring bands, decisions, and EXIF keyword rules.
+
+Pure, deterministic functions extracted from the donors so the model adapter and
+exiftool adapter stay at the edges. This is the shared source of truth for how a
+score becomes a decision and how safety keywords are read and projected.
+
+Extracted from (donor_ledger.yaml):
+- pa-nsfw-filter — photo_analyzer._rec_has_nsfw / filter_nsfw_tagged
+- nt-exif-marks  — nsfwtag.exif.read_marks / read_tagged (keyword parsing)
+- nsfwtag thresholds (DEFAULT_THRESHOLD / DEFAULT_REVIEW_MIN)
+
+Behavior changes vs the donors (required by stable identity + shared jobs):
+- Decisions and scores key on ``asset_id`` and live in the database, not in a
+  path-keyed ``nsfw_scores.csv`` (the CSV is a one-time migration import).
+- Safety keywords are **mutually exclusive**: writing a decision adds the chosen
+  ``sfw``/``nsfw`` keyword and removes the opposite. The donor only ever added
+  ``nsfw``; there was no ``sfw`` keyword and no removal.
+- Model scoring runs behind ``integrations.nsfw_model.NsfwModel`` as a shared job,
+  not inline in a CLI pass.
+- No module here imports the archived CLIs (``nsfwtag`` / ``photo_analyzer``).
+"""
+
+from __future__ import annotations
+
+from collections.abc import Iterable, Mapping
+
+NSFW = "nsfw"
+SFW = "sfw"
+REVIEW_REQUIRED = "review_required"
+
+DEFAULT_THRESHOLD = 0.6  # AdamCodd model sweet spot (donor DEFAULT_THRESHOLD)
+DEFAULT_REVIEW_MIN = 0.2  # below this is confidently SFW (donor DEFAULT_REVIEW_MIN)
+
+
+def classify(
+    score: float, *, threshold: float = DEFAULT_THRESHOLD, review_min: float = DEFAULT_REVIEW_MIN
+) -> str:
+    """Suggested decision from a model score. A human confirms before it is durable.
+
+    ``>= threshold`` → ``nsfw``; ``>= review_min`` → ``review_required``; else ``sfw``.
+    """
+    if score >= threshold:
+        return NSFW
+    if score >= review_min:
+        return REVIEW_REQUIRED
+    return SFW
+
+
+def normalize_keywords(values: Iterable | str | None) -> set[str]:
+    """Normalize an EXIF Keywords/Subject value (list or scalar) to a lowercased set."""
+    if values is None:
+        return set()
+    if isinstance(values, str):
+        values = [values]
+    return {str(v).strip().lower() for v in values}
+
+
+def has_nsfw(keywords: Iterable[str]) -> bool:
+    """True if the ``nsfw`` safety keyword is present (donor _rec_has_nsfw)."""
+    return NSFW in {str(k).strip().lower() for k in keywords}
+
+
+def marks_from_keywords(keyword_sets: Mapping[str, Iterable[str]]) -> dict[str, set[str]]:
+    """Split ``{path: keywords}`` into ``{'nsfw': set, 'sfw': set}``.
+
+    They should be mutually exclusive; if a file carries both, ``nsfw`` wins so it
+    never reads as safe (donor nsfwtag.exif.read_marks).
+    """
+    nsfw, sfw = set(), set()
+    for path, keywords in keyword_sets.items():
+        lowered = {str(k).strip().lower() for k in keywords}
+        if NSFW in lowered:
+            nsfw.add(path)
+        elif SFW in lowered:
+            sfw.add(path)
+    return {NSFW: nsfw, SFW: sfw}
+
+
+def partition_nsfw(
+    keyword_sets: Mapping[str, Iterable[str]],
+) -> tuple[list[str], list[str]]:
+    """Split paths into ``(analyzable, skipped)`` — nsfw-tagged files skip cloud
+    analysis (privacy gate; donor photo_analyzer.filter_nsfw_tagged)."""
+    analyzable, skipped = [], []
+    for path, keywords in keyword_sets.items():
+        (skipped if has_nsfw(keywords) else analyzable).append(path)
+    return analyzable, skipped
+
+
+def exif_projection(decision: str) -> dict[str, list[str]]:
+    """Keyword operations to make EXIF match a confirmed decision.
+
+    Enforces mutual exclusion: add the decision keyword, remove the opposite.
+    Non-terminal decisions (review/unknown) touch nothing.
+    """
+    if decision == NSFW:
+        return {"add": [NSFW], "remove": [SFW]}
+    if decision == SFW:
+        return {"add": [SFW], "remove": [NSFW]}
+    return {"add": [], "remove": []}

tests/e2e/test_traceability.py

@@ -1,21 +1,21 @@
-"""Story-to-test traceability for Phase A (Epic E01).
+"""Story-to-test traceability.
 
-Every story US01-01..US01-07 must map to test files that exist, and every Phase A
-test file must be claimed by a story — so a new test can't go unexercised and a
-story can't quietly lose its coverage. Whether those tests *pass* is proven by
-running the suite; this guards the mapping's completeness.
+Every mapped story must map to test files that exist, every test file must be
+claimed by a story (so a new test can't go unexercised), and all Phase A stories
+US01-01..US01-07 must be present. Whether those tests *pass* is proven by running
+the suite; this guards the mapping's completeness.
 """
 
 import json
 from pathlib import Path
 
 REPO = Path(__file__).resolve().parents[2]
-MAP = json.loads((REPO / "tests" / "phase_a_traceability.json").read_text())["stories"]
-EXPECTED_STORIES = {f"US01-0{n}" for n in range(1, 8)}
+MAP = json.loads((REPO / "tests" / "story_traceability.json").read_text())["stories"]
+PHASE_A_STORIES = {f"US01-0{n}" for n in range(1, 8)}
 
 
 def test_all_phase_a_stories_are_mapped():
-    assert set(MAP) == EXPECTED_STORIES
+    assert PHASE_A_STORIES <= set(MAP)
 
 
 def test_every_mapped_test_file_exists_and_is_nonempty():

tests/integration/test_safety_parity.py

@@ -0,0 +1,55 @@
+"""Characterization parity: the extracted safety logic matches the donor on the
+same fixture files (real exiftool-written keywords)."""
+
+import shutil
+import subprocess
+
+import numpy as np
+import pytest
+from PIL import Image
+
+from photo_pipeline.integrations import exiftool
+from photo_pipeline.services import safety
+
+EXIFTOOL = shutil.which("exiftool")
+pytestmark = pytest.mark.skipif(EXIFTOOL is None, reason="exiftool not installed")
+
+
+def _jpeg(path, seed=1):
+    path.parent.mkdir(parents=True, exist_ok=True)
+    arr = np.random.default_rng(seed).integers(0, 256, (32, 32, 3), dtype=np.uint8)
+    Image.fromarray(arr).save(path, quality=90)
+    return str(path)
+
+
+def _tag(path, keyword):
+    subprocess.run(
+        ["exiftool", "-m", "-overwrite_original", f"-Keywords={keyword}", f"-Subject={keyword}", path],
+        check=True,
+        capture_output=True,
+    )
+
+
+def test_extracted_marks_and_partition_match_donor(tmp_path):
+    donor_exif = pytest.importorskip("nsfwtag.exif")
+
+    nsfw = _jpeg(tmp_path / "nsfw.jpg", 1)
+    sfw = _jpeg(tmp_path / "sfw.jpg", 2)
+    plain = _jpeg(tmp_path / "plain.jpg", 3)
+    _tag(nsfw, "nsfw")
+    _tag(sfw, "sfw")
+    paths = [nsfw, sfw, plain]
+
+    # Donor behavior.
+    donor_marks = donor_exif.read_marks(paths)
+    donor_nsfw = donor_exif.read_tagged(paths, "nsfw")
+
+    # Extracted behavior on the same files.
+    keyword_sets = exiftool.read_keyword_sets(paths)
+    extracted_marks = safety.marks_from_keywords(keyword_sets)
+    _, skipped = safety.partition_nsfw(keyword_sets)
+
+    assert extracted_marks == donor_marks
+    assert set(skipped) == donor_nsfw
+    assert extracted_marks["nsfw"] == {nsfw}
+    assert extracted_marks["sfw"] == {sfw}

tests/story_traceability.json

@@ -1,5 +1,5 @@
 {
-  "_comment": "Maps each Phase A (Epic E01) story to the automated tests that exercise it. Verified by tests/e2e/test_traceability.py: every story must map to existing test files, and every Phase A test file must be claimed by a story (no unexercised tests).",
+  "_comment": "Maps each delivered story to the automated tests that exercise it. Verified by tests/e2e/test_traceability.py: every mapped story maps to existing test files, every test file is claimed by a story (no unexercised tests), and all Phase A (E01) stories US01-01..US01-07 are present.",
   "stories": {
     "US01-01": [
       "tests/characterization/test_donor_ledger.py",
@@ -39,6 +39,11 @@
     "US01-07": [
       "tests/e2e/test_phase_a_pipeline.py",
       "tests/e2e/test_traceability.py"
+    ],
+    "US02-01": [
+      "tests/unit/test_safety.py",
+      "tests/unit/test_design_tokens.py",
+      "tests/integration/test_safety_parity.py"
     ]
   }
 }

tests/unit/test_design_tokens.py

@@ -0,0 +1,22 @@
+"""Snapshot of the shared design tokens extracted from the donor UIs."""
+
+import re
+from pathlib import Path
+
+REPO = Path(__file__).resolve().parents[2]
+TOKENS = REPO / "frontend" / "css" / "tokens.css"
+
+EXPECTED = {
+    "--bg", "--surface", "--surface-2", "--border", "--text", "--muted",
+    "--accent", "--danger", "--ok", "--warn", "--radius",
+}
+
+
+def test_token_set_matches_snapshot():
+    names = set(re.findall(r"(--[a-z0-9-]+):", TOKENS.read_text()))
+    assert names == EXPECTED  # changing tokens is intentional; update this snapshot
+
+
+def test_app_css_imports_the_shared_tokens():
+    app_css = (REPO / "frontend" / "css" / "app.css").read_text()
+    assert '@import "./tokens.css";' in app_css

tests/unit/test_safety.py

@@ -0,0 +1,68 @@
+"""Extracted safety logic: scoring bands, keyword rules, and no CLI dependency."""
+
+import ast
+from pathlib import Path
+
+import pytest
+
+from photo_pipeline.services import safety
+
+REPO = Path(__file__).resolve().parents[2]
+
+
+@pytest.mark.parametrize(
+    "score,expected",
+    [(0.95, "nsfw"), (0.6, "nsfw"), (0.59, "review_required"), (0.2, "review_required"), (0.05, "sfw")],
+)
+def test_classify_bands(score, expected):
+    assert safety.classify(score) == expected
+
+
+def test_normalize_keywords_handles_list_scalar_and_none():
+    assert safety.normalize_keywords(["NSFW", " Sfw "]) == {"nsfw", "sfw"}
+    assert safety.normalize_keywords("NSFW") == {"nsfw"}
+    assert safety.normalize_keywords(None) == set()
+
+
+def test_has_nsfw():
+    assert safety.has_nsfw(["Foo", "NSFW"])
+    assert not safety.has_nsfw(["sfw", "beach"])
+
+
+def test_marks_nsfw_wins_over_sfw():
+    marks = safety.marks_from_keywords(
+        {"a.jpg": ["nsfw", "sfw"], "b.jpg": ["sfw"], "c.jpg": ["landscape"]}
+    )
+    assert marks == {"nsfw": {"a.jpg"}, "sfw": {"b.jpg"}}
+
+
+def test_partition_nsfw_gates_analysis():
+    analyzable, skipped = safety.partition_nsfw({"keep.jpg": ["x"], "block.jpg": ["nsfw"]})
+    assert analyzable == ["keep.jpg"]
+    assert skipped == ["block.jpg"]
+
+
+def test_exif_projection_is_mutually_exclusive():
+    assert safety.exif_projection("nsfw") == {"add": ["nsfw"], "remove": ["sfw"]}
+    assert safety.exif_projection("sfw") == {"add": ["sfw"], "remove": ["nsfw"]}
+    assert safety.exif_projection("review_required") == {"add": [], "remove": []}
+
+
+def test_shared_modules_do_not_import_archived_clis():
+    """Acceptance: shared modules have no dependency on archived entry points."""
+    banned = {"nsfwtag", "photo_analyzer", "nsfw_tag"}
+    modules = [
+        REPO / "photo_pipeline" / "services" / "safety.py",
+        REPO / "photo_pipeline" / "integrations" / "exiftool.py",
+        REPO / "photo_pipeline" / "integrations" / "nsfw_model.py",
+    ]
+    for module in modules:
+        tree = ast.parse(module.read_text(), filename=str(module))
+        for node in ast.walk(tree):
+            if isinstance(node, ast.Import):
+                names = {a.name.split(".")[0] for a in node.names}
+            elif isinstance(node, ast.ImportFrom):
+                names = {(node.module or "").split(".")[0]}
+            else:
+                continue
+            assert not (names & banned), f"{module.name} imports archived CLI: {names & banned}"