US01-05: Generate and Serve Managed Thumbnails

migrations/versions/0003_thumbnails.py

@@ -0,0 +1,46 @@
+"""Managed thumbnail cache table.
+
+Revision ID: 0003_thumbnails
+Revises: 0002_duplicate_engine
+Create Date: 2026-07-15
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = "0003_thumbnails"
+down_revision = "0002_duplicate_engine"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "thumbnails",
+        sa.Column("cache_key", sa.String(), primary_key=True),
+        sa.Column("asset_id", sa.String(), sa.ForeignKey("assets.id"), nullable=False),
+        sa.Column("size", sa.Integer(), nullable=False),
+        sa.Column("state", sa.String(), nullable=False),
+        sa.Column("error_code", sa.String(), nullable=True),
+        sa.Column("path", sa.String(), nullable=True),
+        sa.Column("width", sa.Integer(), nullable=True),
+        sa.Column("height", sa.Integer(), nullable=True),
+        sa.Column("format", sa.String(), nullable=True),
+        sa.Column(
+            "created_at",
+            sa.DateTime(timezone=True),
+            nullable=False,
+            server_default=sa.text("CURRENT_TIMESTAMP"),
+        ),
+        sa.Column(
+            "updated_at",
+            sa.DateTime(timezone=True),
+            nullable=False,
+            server_default=sa.text("CURRENT_TIMESTAMP"),
+        ),
+    )
+    op.create_index("ix_thumbnails_asset_id", "thumbnails", ["asset_id"])
+
+
+def downgrade() -> None:
+    op.drop_table("thumbnails")

photo_pipeline/api/app.py

@@ -12,7 +12,7 @@ from contextlib import asynccontextmanager
 
 from fastapi import FastAPI
 
-from photo_pipeline.api.routes import health
+from photo_pipeline.api.routes import health, thumbnails
 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
@@ -38,4 +38,5 @@ def create_app(config: Config | None = None) -> FastAPI:
 
     app = FastAPI(title="Photo Pipeline", version="0.1.0", lifespan=lifespan)
     app.include_router(health.router, prefix="/api/v1")
+    app.include_router(thumbnails.router, prefix="/api/v1")
     return app

photo_pipeline/api/routes/thumbnails.py

@@ -0,0 +1,35 @@
+"""Thumbnail endpoint: previews addressed only by asset ID and bounded size.
+
+The browser never supplies a filesystem path. Errors return a consistent JSON
+envelope with the service's typed code and status; successful responses carry the
+image with an immutable cache header (the URL's identity comes from the versioned,
+content-keyed cache).
+"""
+
+from __future__ import annotations
+
+from fastapi import APIRouter, Query, Request
+from fastapi.responses import FileResponse, JSONResponse
+
+from photo_pipeline.services.thumbnails import ThumbnailError, ThumbnailService
+
+router = APIRouter(tags=["thumbnails"])
+
+
+@router.get("/assets/{asset_id}/thumbnail")
+def get_thumbnail(asset_id: str, request: Request, size: int = Query(512)):
+    service = ThumbnailService(
+        request.app.state.session_factory, request.app.state.config
+    )
+    try:
+        path = service.generate(asset_id, size)
+    except ThumbnailError as error:
+        return JSONResponse(
+            status_code=error.http_status,
+            content={"error": {"code": error.code, "message": str(error)}},
+        )
+    return FileResponse(
+        path,
+        media_type="image/webp",
+        headers={"Cache-Control": "public, max-age=31536000, immutable"},
+    )

photo_pipeline/config.py

@@ -30,6 +30,11 @@ class Config(BaseModel):
     log_level: str = "INFO"
     log_format: str = "json"  # "json" or "text"
 
+    # Library boundary for path validation (os.pathsep-separated in the env var).
+    library_roots: tuple[Path, ...] = ()
+    thumbnail_cache_quota_bytes: int = 500_000_000
+    thumbnail_max_pixels: int = 100_000_000
+
     vision_api_key: SecretStr | None = None
     immich_api_key: SecretStr | None = None
 
@@ -41,12 +46,17 @@ class Config(BaseModel):
     def database_url(self) -> str:
         return f"sqlite:///{self.database_path}"
 
+    @property
+    def thumbnail_cache_dir(self) -> Path:
+        return self.data_dir / "cache" / "thumbs"
+
     @classmethod
     def from_env(cls, environ: Mapping[str, str] | None = None) -> "Config":
         env = os.environ if environ is None else environ
-        data = {
+        data: dict = {}
-            name: env[ENV_PREFIX + name.upper()]
+        for name in cls.model_fields:
-            for name in cls.model_fields
+            raw = env.get(ENV_PREFIX + name.upper())
-            if env.get(ENV_PREFIX + name.upper())
+            if not raw:
-        }
+                continue
+            data[name] = raw.split(os.pathsep) if name == "library_roots" else raw
         return cls(**data)

photo_pipeline/models/__init__.py

@@ -10,6 +10,7 @@ from photo_pipeline.models.duplicates import (
     DuplicateMember,
     DuplicateNegativeLink,
 )
+from photo_pipeline.models.thumbnails import Thumbnail
 
 __all__ = [
     "Asset",
@@ -17,4 +18,5 @@ __all__ = [
     "DuplicateCluster",
     "DuplicateMember",
     "DuplicateNegativeLink",
+    "Thumbnail",
 ]

photo_pipeline/models/thumbnails.py

@@ -0,0 +1,37 @@
+"""Managed thumbnail cache records.
+
+One row per ``cache_key`` (pixel hash + size + algorithm version, or the source
+byte hash for undecodable inputs). The row records whether generation succeeded or
+failed and why, so a broken original is not retried on every request. Because the
+key contains the normalized pixel hash, an EXIF-only change reuses the thumbnail
+and a genuine pixel change invalidates it automatically.
+"""
+
+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 Thumbnail(Base):
+    __tablename__ = "thumbnails"
+
+    cache_key: Mapped[str] = mapped_column(String, primary_key=True)
+    asset_id: Mapped[str] = mapped_column(ForeignKey("assets.id"), nullable=False)
+    size: Mapped[int] = mapped_column(Integer, nullable=False)
+    state: Mapped[str] = mapped_column(String, nullable=False)  # ready | error
+    error_code: Mapped[str | None] = mapped_column(String)
+    path: Mapped[str | None] = mapped_column(String)
+    width: Mapped[int | None] = mapped_column(Integer)
+    height: Mapped[int | None] = mapped_column(Integer)
+    format: Mapped[str | None] = mapped_column(String)
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, server_default=func.now()
+    )
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now()
+    )

photo_pipeline/services/thumbnails.py

@@ -0,0 +1,274 @@
+"""ThumbnailService — safe, cached, correctly oriented previews by asset ID.
+
+Thumbnails are requested by asset ID (never a browser-supplied path), resolved to
+the current path through the database, and validated against the configured library
+boundary. Generation applies EXIF orientation before resizing, preserves
+transparency, and encodes WebP. The cache key is ``pixel_hash + size + version``, so
+an EXIF-only edit reuses the file while a real pixel change invalidates it; writes
+are atomic and the cache is bounded by an LRU quota. Failures are persisted as typed
+errors so a broken original is not retried on every request.
+
+Reuses photo_analyzer.prepare_image decode/resize/HEIC handling, adding the missing
+EXIF-orientation step, WebP output, and a managed cache (donor_ledger.yaml:
+pa-imaging).
+"""
+
+from __future__ import annotations
+
+import uuid
+from pathlib import Path
+
+from PIL import Image, ImageOps, UnidentifiedImageError
+from sqlalchemy import select
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.orm import sessionmaker
+
+from photo_pipeline import path_policy
+from photo_pipeline.config import Config
+from photo_pipeline.models import Asset, Thumbnail
+
+# Best-effort HEIC support: registered only if the optional decoder is installed.
+try:  # pragma: no cover - depends on an optional native dependency
+    import pillow_heif
+
+    pillow_heif.register_heif_opener()
+except Exception:  # pragma: no cover
+    pass
+
+SIZES = (256, 512, 1280)
+THUMB_VERSION = 1
+THUMB_FORMAT = "webp"
+
+
+class ThumbnailError(RuntimeError):
+    code = "thumbnail_error"
+    http_status = 500
+
+
+class InvalidSize(ThumbnailError):
+    code = "invalid_size"
+    http_status = 422
+
+
+class ThumbnailNotFound(ThumbnailError):
+    code = "not_found"
+    http_status = 404
+
+
+class ThumbnailUnavailable(ThumbnailError):
+    code = "unavailable"
+    http_status = 409
+
+
+class UnsupportedImage(ThumbnailError):
+    code = "unsupported_image"
+    http_status = 415
+
+
+class ImageTooLarge(ThumbnailError):
+    code = "image_too_large"
+    http_status = 413
+
+
+class PathNotAllowed(ThumbnailError):
+    code = "path_not_allowed"
+    http_status = 403
+
+
+# Errors that describe a broken source and are persisted to avoid retry loops.
+_PERSISTED_ERRORS = {UnsupportedImage, ImageTooLarge}
+_ERROR_BY_CODE = {cls.code: cls for cls in _PERSISTED_ERRORS}
+
+
+class ThumbnailService:
+    def __init__(self, session_factory: sessionmaker, config: Config) -> None:
+        self._session_factory = session_factory
+        self._config = config
+        self._cache_dir = config.thumbnail_cache_dir
+
+    def generate(self, asset_id: str, size: int) -> Path:
+        if size not in SIZES:
+            raise InvalidSize(f"size must be one of {SIZES}")
+
+        with self._session_factory() as session:
+            asset = session.get(Asset, asset_id)
+            if asset is None:
+                raise ThumbnailNotFound(f"unknown asset {asset_id}")
+            if asset.availability_state != "active" or not asset.current_path:
+                raise ThumbnailUnavailable(f"asset {asset_id} has no active file")
+            self._validate_path(asset.current_path)
+            cache_key = self._cache_key(asset, size)
+
+            row = session.get(Thumbnail, cache_key)
+            if row is not None:
+                if row.state == "error":
+                    raise _ERROR_BY_CODE[row.error_code](
+                        f"cached failure for {asset_id}: {row.error_code}"
+                    )
+                if row.path and Path(row.path).exists():
+                    _touch(row.path)
+                    return Path(row.path)
+
+            source = asset.current_path
+
+        # Rendering happens outside the DB session (no transaction held during I/O).
+        try:
+            rendered = self._render(source, size, cache_key)
+        except ThumbnailError as error:
+            if type(error) in _PERSISTED_ERRORS:
+                self._record_error(cache_key, asset_id, size, error.code)
+            raise
+
+        self._record_ready(cache_key, asset_id, size, rendered)
+        self._enforce_quota(keep=rendered["path"])
+        return Path(rendered["path"])
+
+    # ── path safety ──────────────────────────────────────────────────────────
+    def _validate_path(self, current_path: str) -> None:
+        path = Path(current_path)
+        if path_policy.is_excluded(path):
+            raise PathNotAllowed(f"excluded path: {current_path}")
+        roots = self._config.library_roots
+        if not roots:
+            return
+        for root in roots:
+            try:
+                path_policy.resolve_within(Path(root), path)
+                return
+            except path_policy.PathPolicyError:
+                continue
+        raise PathNotAllowed(f"path outside configured roots: {current_path}")
+
+    # ── cache key + rendering ──────────────────────────────────────────────────
+    @staticmethod
+    def _cache_key(asset: Asset, size: int) -> str:
+        if asset.pixel_sha256:
+            return f"{asset.pixel_sha256}-{size}-v{THUMB_VERSION}"
+        # Undecodable at inventory time: key by source bytes so a fixed file retries.
+        return f"src:{asset.current_sha256 or asset.id}-{size}-v{THUMB_VERSION}"
+
+    def _cache_path(self, cache_key: str) -> Path:
+        safe = cache_key.replace(":", "_")
+        return self._cache_dir / safe[:2] / f"{safe}.{THUMB_FORMAT}"
+
+    def _render(self, source: str, size: int, cache_key: str) -> dict:
+        try:
+            with Image.open(source) as image:
+                width, height = image.size
+                if width * height > self._config.thumbnail_max_pixels:
+                    raise ImageTooLarge(
+                        f"{width}x{height} exceeds {self._config.thumbnail_max_pixels} px"
+                    )
+                oriented = ImageOps.exif_transpose(image)
+                mode = "RGBA" if _has_alpha(oriented) else "RGB"
+                converted = oriented.convert(mode)
+                converted.thumbnail((size, size), Image.LANCZOS)
+                out_width, out_height = converted.size
+                destination = self._cache_path(cache_key)
+                destination.parent.mkdir(parents=True, exist_ok=True)
+                tmp = destination.with_name(f".{destination.name}.{uuid.uuid4().hex}.tmp")
+                converted.save(tmp, format="WEBP", quality=82, method=4)
+        except ImageTooLarge:
+            raise
+        except (UnidentifiedImageError, OSError, ValueError) as error:
+            raise UnsupportedImage(f"cannot decode {source}: {error}") from error
+
+        import os
+
+        os.replace(tmp, destination)
+        return {
+            "path": str(destination),
+            "width": out_width,
+            "height": out_height,
+            "format": THUMB_FORMAT,
+        }
+
+    # ── persistence ────────────────────────────────────────────────────────────
+    def _record_ready(self, cache_key: str, asset_id: str, size: int, rendered: dict) -> None:
+        with self._session_factory() as session:
+            session.merge(
+                Thumbnail(
+                    cache_key=cache_key,
+                    asset_id=asset_id,
+                    size=size,
+                    state="ready",
+                    error_code=None,
+                    path=rendered["path"],
+                    width=rendered["width"],
+                    height=rendered["height"],
+                    format=rendered["format"],
+                )
+            )
+            try:
+                session.commit()
+            except IntegrityError:
+                session.rollback()  # another worker recorded it first; the file stands
+
+    def _record_error(self, cache_key: str, asset_id: str, size: int, code: str) -> None:
+        with self._session_factory() as session:
+            session.merge(
+                Thumbnail(
+                    cache_key=cache_key,
+                    asset_id=asset_id,
+                    size=size,
+                    state="error",
+                    error_code=code,
+                )
+            )
+            try:
+                session.commit()
+            except IntegrityError:
+                session.rollback()
+
+    # ── quota / eviction ───────────────────────────────────────────────────────
+    def _enforce_quota(self, keep: str | None = None) -> None:
+        quota = self._config.thumbnail_cache_quota_bytes
+        if not self._cache_dir.exists():
+            return
+        files = [f for f in self._cache_dir.rglob(f"*.{THUMB_FORMAT}") if f.is_file()]
+        total = sum(f.stat().st_size for f in files)
+        if total <= quota:
+            return
+        files.sort(key=lambda f: f.stat().st_mtime)  # least-recently-used first
+        keep_path = str(Path(keep)) if keep else None
+        evicted: list[str] = []
+        for f in files:
+            if total <= quota:
+                break
+            if str(f) == keep_path:
+                continue
+            size = f.stat().st_size
+            try:
+                f.unlink()
+                total -= size
+                evicted.append(str(f))
+            except OSError:
+                pass
+        if evicted:
+            self._forget(evicted)
+
+    def _forget(self, paths: list[str]) -> None:
+        with self._session_factory() as session:
+            rows = session.execute(
+                select(Thumbnail).where(Thumbnail.path.in_(paths))
+            ).scalars()
+            for row in rows:
+                session.delete(row)
+            session.commit()
+
+
+def _has_alpha(image: Image.Image) -> bool:
+    return image.mode in ("RGBA", "LA") or (
+        image.mode == "P" and "transparency" in image.info
+    )
+
+
+def _touch(path: str) -> None:
+    import os
+    import time
+
+    try:
+        now = time.time()
+        os.utime(path, (now, now))
+    except OSError:
+        pass

tests/integration/test_thumbnails.py

@@ -0,0 +1,249 @@
+"""Thumbnail service + endpoint: orientation, transparency, caching, concurrency,
+corruption, invalidation, eviction, and path safety.
+
+Fixtures are inlined (not in a sibling conftest.py) to avoid shadowing the
+characterization suite's bare ``conftest`` import under pytest's prepend mode.
+"""
+
+from concurrent.futures import ThreadPoolExecutor
+from datetime import datetime, timezone
+from types import SimpleNamespace
+
+import numpy as np
+import pytest
+from fastapi.testclient import TestClient
+from PIL import Image
+from sqlalchemy import select
+
+from photo_pipeline.api.app import create_app
+from photo_pipeline.config import Config
+from photo_pipeline.db import create_db_engine, create_session_factory, run_migrations
+from photo_pipeline.models import Asset, Thumbnail
+from photo_pipeline.services.inventory import InventoryService
+from photo_pipeline.services.thumbnails import (
+    ImageTooLarge,
+    InvalidSize,
+    PathNotAllowed,
+    ThumbnailNotFound,
+    ThumbnailService,
+    UnsupportedImage,
+)
+
+
+@pytest.fixture
+def env(tmp_path):
+    data = tmp_path / "data"
+    data.mkdir()
+    lib = tmp_path / "lib"
+    lib.mkdir()
+    config = Config.from_env(
+        {
+            "PHOTO_PIPELINE_DATA_DIR": str(data),
+            "PHOTO_PIPELINE_LIBRARY_ROOTS": str(lib),
+        }
+    )
+    run_migrations(config.database_url)
+    engines = []
+
+    def factory():
+        engine = create_db_engine(config.database_url)
+        engines.append(engine)
+        return create_session_factory(engine)
+
+    sf = factory()
+    yield SimpleNamespace(config=config, lib=lib, data=data, sf=sf, factory=factory)
+    for engine in engines:
+        engine.dispose()
+
+
+@pytest.fixture
+def thumbs(env):
+    return ThumbnailService(env.sf, env.config)
+
+
+@pytest.fixture
+def inventory(env):
+    return InventoryService(env.sf)
+
+
+def plain_jpeg(path, seed, size=(200, 150)):
+    path.parent.mkdir(parents=True, exist_ok=True)
+    arr = np.random.default_rng(seed).integers(0, 256, (size[1], size[0], 3), dtype=np.uint8)
+    Image.fromarray(arr).save(path, quality=90)
+    return path
+
+
+def oriented_jpeg(path, w, h, orientation, seed=1):
+    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  # 274 = Orientation
+    img.save(path, exif=exif, quality=90)
+    return path
+
+
+def transparent_png(path, size=(120, 90)):
+    arr = np.random.default_rng(2).integers(0, 256, (size[1], size[0], 4), dtype=np.uint8)
+    Image.fromarray(arr, "RGBA").save(path)
+    return path
+
+
+def first_asset_id(scan_result):
+    return next(iter(scan_result.asset_ids.values()))
+
+
+def insert_asset(factory, asset_id, current_path, pixel="pix", sha="sha"):
+    with factory() as session:
+        session.add(
+            Asset(
+                id=asset_id,
+                original_path=current_path,
+                current_path=current_path,
+                current_sha256=sha,
+                pixel_sha256=pixel,
+                hash_version=1,
+                discovered_at=datetime.now(timezone.utc),
+                availability_state="active",
+            )
+        )
+        session.commit()
+
+
+def test_applies_orientation_and_downsizes(env, thumbs, inventory):
+    # 1000x600 landscape tagged "rotate" -> displayed 600x1000 portrait.
+    oriented_jpeg(env.lib / "rot.jpg", 1000, 600, orientation=6)
+    asset_id = first_asset_id(inventory.scan(env.lib))
+
+    path = thumbs.generate(asset_id, 512)
+    with Image.open(path) as image:
+        assert image.format == "WEBP"
+        assert max(image.size) == 512  # long edge bounded by requested size
+        assert image.height > image.width  # orientation was applied before resize
+
+
+def test_preserves_transparency(env, thumbs, inventory):
+    transparent_png(env.lib / "t.png")
+    asset_id = first_asset_id(inventory.scan(env.lib))
+    path = thumbs.generate(asset_id, 256)
+    with Image.open(path) as image:
+        assert "A" in image.getbands()  # alpha survived into the WebP
+
+
+def test_cache_hit_returns_same_file(env, thumbs, inventory):
+    plain_jpeg(env.lib / "a.jpg", 1)
+    asset_id = first_asset_id(inventory.scan(env.lib))
+    first = thumbs.generate(asset_id, 512)
+    second = thumbs.generate(asset_id, 512)
+    assert first == second
+
+
+def test_concurrent_generation_yields_one_valid_file(env, thumbs, inventory):
+    plain_jpeg(env.lib / "a.jpg", 1, size=(800, 600))
+    asset_id = first_asset_id(inventory.scan(env.lib))
+
+    with ThreadPoolExecutor(max_workers=8) as pool:
+        paths = list(pool.map(lambda _: str(thumbs.generate(asset_id, 512)), range(8)))
+
+    assert len(set(paths)) == 1
+    with Image.open(paths[0]) as image:
+        assert image.format == "WEBP"
+
+
+def test_corrupt_source_errors_and_is_not_retried(env, thumbs, inventory):
+    (env.lib / "broken.jpg").write_bytes(b"this is not an image")
+    asset_id = first_asset_id(inventory.scan(env.lib))
+
+    with pytest.raises(UnsupportedImage):
+        thumbs.generate(asset_id, 512)
+
+    with env.factory()() as session:
+        rows = list(session.execute(select(Thumbnail)).scalars())
+    assert rows and rows[0].state == "error"
+
+    # The persisted failure is replayed rather than re-rendered.
+    with pytest.raises(UnsupportedImage):
+        thumbs.generate(asset_id, 512)
+
+
+def test_pixel_change_invalidates_cache(env, thumbs, inventory):
+    a = plain_jpeg(env.lib / "a.jpg", 1)
+    asset_id = first_asset_id(inventory.scan(env.lib))
+    before = thumbs.generate(asset_id, 512)
+
+    plain_jpeg(a, seed=99)  # different pixels at the same path
+    inventory.scan(env.lib)  # refreshes pixel_sha256
+    after = thumbs.generate(asset_id, 512)
+
+    assert before != after  # new pixel identity -> new cache key
+
+
+def test_quota_evicts_least_recently_used(env, inventory):
+    for i in range(3):
+        plain_jpeg(env.lib / f"{i}.jpg", seed=i, size=(400, 300))
+    ids = list(inventory.scan(env.lib).asset_ids.values())
+
+    tiny = env.config.model_copy(update={"thumbnail_cache_quota_bytes": 100})
+    service = ThumbnailService(env.sf, tiny)
+    last = None
+    for asset_id in ids:
+        last = service.generate(asset_id, 512)
+
+    remaining = list((env.config.thumbnail_cache_dir).rglob("*.webp"))
+    assert len(remaining) < 3  # eviction happened
+    assert last is not None and last.exists()  # the just-created file is pinned
+
+
+def test_path_outside_roots_is_denied(env, thumbs, tmp_path):
+    outside = tmp_path / "outside.jpg"
+    plain_jpeg(outside, 1)
+    insert_asset(env.sf, "a-out", str(outside))
+    with pytest.raises(PathNotAllowed):
+        thumbs.generate("a-out", 512)
+
+
+def test_excluded_path_is_denied(env, thumbs):
+    hidden = env.lib / "_IGNORE" / "secret.jpg"
+    plain_jpeg(hidden, 1)
+    insert_asset(env.sf, "a-ign", str(hidden))
+    with pytest.raises(PathNotAllowed):
+        thumbs.generate("a-ign", 512)
+
+
+def test_unknown_asset_and_bad_size(env, thumbs):
+    with pytest.raises(ThumbnailNotFound):
+        thumbs.generate("ghost", 512)
+    with pytest.raises(InvalidSize):
+        thumbs.generate("ghost", 999)
+
+
+def test_oversized_image_is_rejected(env, inventory):
+    plain_jpeg(env.lib / "big.jpg", 1, size=(400, 300))
+    asset_id = first_asset_id(inventory.scan(env.lib))
+    tiny = env.config.model_copy(update={"thumbnail_max_pixels": 1000})
+    service = ThumbnailService(env.sf, tiny)
+    with pytest.raises(ImageTooLarge):
+        service.generate(asset_id, 512)
+
+
+def test_api_endpoint_serves_and_reports_errors(env):
+    plain_jpeg(env.lib / "a.jpg", 1, size=(800, 600))
+    app = create_app(env.config)
+    with TestClient(app) as client:
+        InventoryService(app.state.session_factory).scan(env.lib)
+        asset_id = next(
+            iter(
+                {
+                    a.id
+                    for a in app.state.session_factory().execute(select(Asset)).scalars()
+                }
+            )
+        )
+        ok = client.get(f"/api/v1/assets/{asset_id}/thumbnail?size=512")
+        assert ok.status_code == 200
+        assert ok.headers["content-type"] == "image/webp"
+        assert "immutable" in ok.headers.get("cache-control", "")
+
+        assert client.get("/api/v1/assets/ghost/thumbnail?size=512").status_code == 404
+        assert (
+            client.get(f"/api/v1/assets/{asset_id}/thumbnail?size=999").status_code == 422
+        )

tests/unit/test_thumbnail_cache_key.py

@@ -0,0 +1,37 @@
+"""Cache-key golden rules: keyed by pixel hash + size + version, so identity
+changes invalidate and EXIF-only changes reuse."""
+
+from photo_pipeline.models import Asset
+from photo_pipeline.services.thumbnails import THUMB_VERSION, ThumbnailService
+
+
+def _asset(pixel=None, sha="deadbeef"):
+    return Asset(
+        id="a1",
+        original_path="a.jpg",
+        current_path="a.jpg",
+        current_sha256=sha,
+        pixel_sha256=pixel,
+        hash_version=1,
+    )
+
+
+def test_key_is_pixel_size_and_version():
+    key = ThumbnailService._cache_key(_asset(pixel="PIX"), 512)
+    assert key == f"PIX-512-v{THUMB_VERSION}"
+
+
+def test_different_size_changes_key():
+    asset = _asset(pixel="PIX")
+    assert ThumbnailService._cache_key(asset, 256) != ThumbnailService._cache_key(asset, 512)
+
+
+def test_different_pixels_change_key():
+    assert ThumbnailService._cache_key(_asset(pixel="P1"), 512) != ThumbnailService._cache_key(
+        _asset(pixel="P2"), 512
+    )
+
+
+def test_undecodable_asset_keys_by_source_bytes():
+    key = ThumbnailService._cache_key(_asset(pixel=None, sha="abc123"), 256)
+    assert key == f"src:abc123-256-v{THUMB_VERSION}"