README.md

@@ -3,3 +3,49 @@
 Integrated, restart-safe photo analysis, duplicate review, metadata, upload, and
 archive workflow. Planning lives in `INTEGRATED_PIPELINE_CONCEPT.md` and
 `delivery_backlog/`.
+
+## Application (`photo_pipeline`)
+
+The target application lives in `photo_pipeline/` (FastAPI + SQLAlchemy + Alembic).
+Run it with:
+
+```bash
+python -m photo_pipeline migrate   # apply database migrations
+python -m photo_pipeline serve     # start the API + static review UI (127.0.0.1:8000)
+```
+
+Configuration comes from `PHOTO_PIPELINE_*` environment variables (see
+`photo_pipeline/config.py`); secrets are referenced, never logged.
+
+## Testing
+
+One offline command runs the whole suite (unit, integration, and browser
+end-to-end); it needs no network and uses only deterministic synthetic fixtures:
+
+```bash
+work_item/scripts/python -m pytest tests -q
+```
+
+Browser end-to-end tests require a one-time Playwright browser install:
+
+```bash
+python -m playwright install chromium
+```
+
+### Phase A acceptance gate
+
+Phase A (Epic E01: shared identity, inventory, duplicates, thumbnails, review UI) is
+gated by a reproducible end-to-end suite:
+
+```bash
+work_item/scripts/python -m pytest tests/e2e tests/integration -q
+```
+
+- `tests/e2e/test_phase_a_pipeline.py` launches the real API process against a fresh
+  database and a deterministic fixture library, then drives scan, `_IGNORE/`
+  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
+  file is left unexercised.

photo_pipeline/api/routes/duplicates.py

@@ -24,6 +24,13 @@ def _error(status: int, code: str, message: str) -> JSONResponse:
     return JSONResponse(status_code=status, content={"error": {"code": code, "message": message}})
 
 
+@router.post("/duplicates/detect")
+def detect(request: Request) -> dict:
+    """Run duplicate detection over the current assets (synchronous in Phase A)."""
+    report = _service(request).detect()
+    return {"clusters": len(report.clusters), "counts": report.counts}
+
+
 @router.get("/duplicates/clusters")
 def list_clusters(
     request: Request,

photo_pipeline/api/routes/inventory.py

@@ -1,14 +1,29 @@
-"""Inventory listing API for the review UI."""
+"""Inventory listing and scan API for the review UI."""
 
 from __future__ import annotations
 
 from fastapi import APIRouter, Query, Request
+from fastapi.responses import JSONResponse
 
 from photo_pipeline.services.inventory import InventoryService
 
 router = APIRouter(tags=["inventory"])
 
 
+@router.post("/inventory/scan")
+def scan(request: Request):
+    """Discover and reconcile the configured library roots (synchronous in Phase A;
+    a durable job in Phase B)."""
+    config = request.app.state.config
+    if not config.library_roots:
+        return JSONResponse(
+            status_code=409,
+            content={"error": {"code": "no_roots", "message": "no library_roots configured"}},
+        )
+    result = InventoryService(request.app.state.session_factory).scan(config.library_roots)
+    return {"counts": result.counts, "total": len(result.asset_ids)}
+
+
 @router.get("/inventory/assets")
 def list_assets(
     request: Request,

tests/e2e/test_phase_a_pipeline.py

@@ -0,0 +1,219 @@
+"""Phase A end-to-end acceptance gate.
+
+Drives the whole Phase A journey against a real API process over HTTP, plus one
+browser reload check, then restarts the process and asserts durable API + database
+state survived. Everything is offline: deterministic synthetic fixtures, no network.
+
+Note: Phase A has no separate worker yet (durable jobs arrive in Phase B/US02); the
+harness launches the real API process, which is the production boundary for this
+phase. Scan and detection run synchronously via their endpoints.
+"""
+
+import shutil
+import socket
+import subprocess
+import sys
+import time
+from io import BytesIO
+from pathlib import Path
+
+import httpx
+import numpy as np
+import pytest
+from PIL import Image
+
+REPO = Path(__file__).resolve().parents[2]
+
+
+def _structured(path, seed, size=(320, 240)):
+    path.parent.mkdir(parents=True, exist_ok=True)
+    rng = np.random.default_rng(seed)
+    w, h = size
+    base = np.zeros((h, w, 3), dtype=np.uint8)
+    for _ in range(6):
+        x0, y0 = int(rng.integers(0, w - 60)), int(rng.integers(0, h - 60))
+        base[y0 : y0 + 60, x0 : x0 + 60] = rng.integers(0, 256, 3)
+    grad = np.linspace(0, 120, w, dtype=np.uint8)
+    base[:, :, 0] = np.clip(base[:, :, 0].astype(int) + grad[None, :], 0, 255)
+    Image.fromarray(base).save(path, quality=95)
+    return path
+
+
+def _resized(src, dst, scale=0.5):
+    with Image.open(src) as image:
+        image.resize(
+            (int(image.width * scale), int(image.height * scale)), Image.LANCZOS
+        ).save(dst, quality=95)
+
+
+def _oriented(path, w, h, orientation=6, seed=7):
+    arr = np.random.default_rng(seed).integers(0, 256, (h, w, 3), dtype=np.uint8)
+    img = Image.fromarray(arr)
+    exif = img.getexif()
+    exif[274] = orientation
+    img.save(path, exif=exif, quality=95)
+
+
+def _free_port():
+    with socket.socket() as sock:
+        sock.bind(("127.0.0.1", 0))
+        return sock.getsockname()[1]
+
+
+class ServerController:
+    def __init__(self, data, lib):
+        self.data = data
+        self.lib = lib
+        self.proc = None
+        self.base = None
+        self.client = None
+
+    def start(self):
+        port = _free_port()
+        env = {
+            "PATH": __import__("os").environ.get("PATH", ""),
+            "PHOTO_PIPELINE_DATA_DIR": str(self.data),
+            "PHOTO_PIPELINE_LIBRARY_ROOTS": str(self.lib),
+            "PHOTO_PIPELINE_HOST": "127.0.0.1",
+            "PHOTO_PIPELINE_PORT": str(port),
+        }
+        self.proc = subprocess.Popen(
+            [sys.executable, "-m", "photo_pipeline", "serve"],
+            cwd=str(REPO),
+            env=env,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+        )
+        self.base = f"http://127.0.0.1:{port}"
+        deadline = time.monotonic() + 30
+        while time.monotonic() < deadline:
+            if self.proc.poll() is not None:
+                _, err = self.proc.communicate()
+                pytest.fail(f"server exited: {err.decode(errors='replace')}")
+            try:
+                if httpx.get(f"{self.base}/api/v1/health/ready", timeout=1).status_code == 200:
+                    self.client = httpx.Client(base_url=self.base, timeout=10)
+                    return
+            except httpx.HTTPError:
+                time.sleep(0.2)
+        self.stop()
+        pytest.fail("server never became ready")
+
+    def restart(self):
+        self.stop()
+        self.start()
+
+    def stop(self):
+        if self.client:
+            self.client.close()
+            self.client = None
+        if self.proc:
+            self.proc.terminate()
+            try:
+                self.proc.wait(timeout=10)
+            except subprocess.TimeoutExpired:
+                self.proc.kill()
+            self.proc = None
+
+
+@pytest.fixture
+def pipeline(tmp_path):
+    data = tmp_path / "data"
+    data.mkdir()
+    lib = tmp_path / "lib"
+    lib.mkdir()
+
+    # Deterministic corpus exercising every Phase A journey.
+    photo = _structured(lib / "album" / "photo.jpg", 1)
+    shutil.copy2(photo, lib / "album" / "photo_copy.jpg")  # exact duplicate
+    scene = _structured(lib / "album" / "scene.jpg", 2)
+    _resized(scene, lib / "album" / "scene_small.jpg")  # perceptual near-duplicate
+    _oriented(lib / "rotated.jpg", 800, 480, orientation=6)  # orientation fixture
+    _structured(lib / "movable.jpg", 5)  # moved between scans
+    _structured(lib / "_IGNORE" / "secret.jpg", 9)  # exclusion sentinel
+
+    controller = ServerController(data, lib)
+    controller.start()
+    yield controller
+    controller.stop()
+
+
+def _assets(client):
+    return client.get("/api/v1/inventory/assets?limit=200").json()["items"]
+
+
+def _id_for(assets, suffix):
+    return next(a["id"] for a in assets if (a["current_path"] or "").endswith(suffix))
+
+
+def test_phase_a_full_pipeline_and_restart(page, pipeline):
+    client = pipeline.client
+
+    # 1. Scan + detect through the real API.
+    assert client.post("/api/v1/inventory/scan").status_code == 200
+    detect = client.post("/api/v1/duplicates/detect").json()
+    assert detect["clusters"] >= 2
+
+    assets = _assets(client)
+    paths = [a["current_path"] for a in assets]
+
+    # 2. Exclusion: the _IGNORE sentinel is never discovered.
+    assert not any("_IGNORE" in p or p.endswith("secret.jpg") for p in paths)
+    # 5 discovered files (photo, photo_copy, scene, scene_small, rotated, movable = 6)
+    assert len(assets) == 6
+    movable_id = _id_for(assets, "movable.jpg")
+    rotated_id = _id_for(assets, "rotated.jpg")
+
+    # 3. Move reconciliation: identity survives a move across a rescan.
+    shutil.move(str(pipeline.lib / "movable.jpg"), str(pipeline.lib / "album" / "moved.jpg"))
+    assert client.post("/api/v1/inventory/scan").status_code == 200
+    assert _id_for(_assets(client), "moved.jpg") == movable_id
+
+    # 4. Exact + fuzzy clusters exist with the right confidence handling.
+    clusters = client.get("/api/v1/duplicates/clusters").json()["items"]
+    exact = next(c for c in clusters if c["method"] == "exact")
+    perceptual = next(c for c in clusters if c["method"] == "perceptual")
+    assert exact["state"] == "decided"  # exact auto-resolved
+    assert perceptual["state"] == "open"  # fuzzy left for review
+
+    # 5. Thumbnail orientation: a rotated source serves an upright preview.
+    thumb = client.get(f"/api/v1/assets/{rotated_id}/thumbnail?size=256")
+    assert thumb.status_code == 200 and thumb.headers["content-type"] == "image/webp"
+    with Image.open(BytesIO(thumb.content)) as image:
+        assert image.height > image.width  # 800x480 landscape shown upright as portrait
+
+    # 6. Canonical selection on the fuzzy cluster persists with variant links.
+    detail = client.get(f"/api/v1/duplicates/clusters/{perceptual['id']}").json()
+    chosen = detail["members"][0]["asset_id"]
+    decided = client.post(
+        f"/api/v1/duplicates/clusters/{perceptual['id']}/decision",
+        json={
+            "decision": "canonical",
+            "expected_version": detail["version"],
+            "canonical_asset_id": chosen,
+        },
+    )
+    assert decided.status_code == 200 and decided.json()["state"] == "decided"
+    variant_ids = {m["asset_id"] for m in detail["members"]} - {chosen}
+    after = {a["id"]: a for a in _assets(client)}
+    assert all(after[v]["canonical_asset_id"] == chosen for v in variant_ids)
+
+    # 7. Reload in the browser: the decision is restored, not re-fetched fresh state.
+    page.goto(f"{pipeline.base}/app/#/duplicates/{perceptual['id']}")
+    page.get_by_test_id("cluster-state").wait_for()
+    page.reload()
+    page.get_by_test_id("cluster-state").wait_for()
+    assert "decided" in page.get_by_test_id("cluster-state").inner_text()
+
+    # 8. Full process restart: durable API + DB state survives.
+    pipeline.restart()
+    client = pipeline.client
+    assert client.get("/api/v1/health/ready").status_code == 200
+    restored = client.get("/api/v1/duplicates/clusters").json()["items"]
+    restored_perceptual = next(c for c in restored if c["id"] == perceptual["id"])
+    assert restored_perceptual["state"] == "decided"  # decision durable
+    restored_assets = {a["id"]: a for a in _assets(client)}
+    assert _id_for(list(restored_assets.values()), "moved.jpg") == movable_id  # identity durable
+    assert all(restored_assets[v]["canonical_asset_id"] == chosen for v in variant_ids)
+    restored_paths = [a["current_path"] for a in restored_assets.values()]
+    assert not any("_IGNORE" in p for p in restored_paths)  # exclusion still holds

tests/e2e/test_traceability.py

@@ -0,0 +1,40 @@
+"""Story-to-test traceability for Phase A (Epic E01).
+
+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.
+"""
+
+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)}
+
+
+def test_all_phase_a_stories_are_mapped():
+    assert set(MAP) == EXPECTED_STORIES
+
+
+def test_every_mapped_test_file_exists_and_is_nonempty():
+    for story, files in MAP.items():
+        assert files, f"{story} maps to no tests"
+        for rel in files:
+            path = REPO / rel
+            assert path.is_file(), f"{story}: missing {rel}"
+            assert path.stat().st_size > 0, f"{story}: empty {rel}"
+            assert "def test_" in path.read_text(), f"{story}: no tests in {rel}"
+
+
+def test_no_unexercised_phase_a_test_files():
+    """Every test_*.py under tests/ is claimed by a story (no orphan coverage)."""
+    mapped = {rel for files in MAP.values() for rel in files}
+    on_disk = {
+        str(p.relative_to(REPO))
+        for p in (REPO / "tests").rglob("test_*.py")
+        if "__pycache__" not in p.parts
+    }
+    unmapped = on_disk - mapped
+    assert not unmapped, f"test files not tied to any story: {sorted(unmapped)}"

tests/phase_a_traceability.json

@@ -0,0 +1,44 @@
+{
+  "_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).",
+  "stories": {
+    "US01-01": [
+      "tests/characterization/test_donor_ledger.py",
+      "tests/characterization/test_pa_discovery.py",
+      "tests/characterization/test_pa_hashing_dedup.py",
+      "tests/characterization/test_pa_imaging.py",
+      "tests/characterization/test_pa_exif.py",
+      "tests/characterization/test_pa_db.py",
+      "tests/characterization/test_pa_album.py",
+      "tests/characterization/test_pa_config.py",
+      "tests/characterization/test_pa_variants.py",
+      "tests/characterization/test_nsfwtag.py",
+      "tests/characterization/test_webapp_query.py"
+    ],
+    "US01-02": [
+      "tests/unit/test_config.py",
+      "tests/integration/test_migrations.py",
+      "tests/integration/test_app_lifecycle.py"
+    ],
+    "US01-03": [
+      "tests/unit/test_path_policy.py",
+      "tests/unit/test_hashing.py",
+      "tests/integration/test_inventory_reconcile.py"
+    ],
+    "US01-04": [
+      "tests/unit/test_phash.py",
+      "tests/integration/test_duplicate_engine.py"
+    ],
+    "US01-05": [
+      "tests/unit/test_thumbnail_cache_key.py",
+      "tests/integration/test_thumbnails.py"
+    ],
+    "US01-06": [
+      "tests/integration/test_review_api.py",
+      "tests/e2e/test_review_ui.py"
+    ],
+    "US01-07": [
+      "tests/e2e/test_phase_a_pipeline.py",
+      "tests/e2e/test_traceability.py"
+    ]
+  }
+}