US07-02: Harden API Authorization and Path Boundaries

.gitignore

@@ -17,6 +17,3 @@ _todo/
 pictures/
 photos/
 _IGNORE/
-
-# Test failure evidence (US07-04)
-.artifacts/

README.md

@@ -228,59 +228,6 @@ work_item/scripts/python -m pytest -m phase_f -q
 
 Phases A–E remain green in the full run above.
 
-## Media and metadata hardening (US07-03)
-
-Every pixel the application reads goes through `photo_pipeline/imaging.py`: the
-declared dimensions are checked before anything is decoded, Pillow's
-decompression-bomb warning is treated as a refusal, JPEG decodes near the requested
-size, and each decoder failure becomes one of two typed errors. A damaged file is a
-per-item error with a persisted code, never a failed scan or a dead worker.
-
-Every metadata stage ends with an EXIF checkpoint (`services/exif_checkpoint.py`):
-snapshot, write the owned keywords, read back, prove the owned fields landed and that
-nothing else moved, refresh the file hash. A field the stage does not own that
-changed anyway makes the checkpoint `divergent` — recorded in `exif_projections`,
-shown in the review queue, never repaired behind the user's back, and not counted as
-verified, so upload stays blocked.
-
-The golden corpus that proves all of it is generated, not committed:
-`tests/fixtures/media_corpus.py` declares every format, orientation, profile,
-damage, and metadata case with its expected outcome, and the suite regenerates it
-twice to prove it does not drift.
-
-```bash
-work_item/scripts/python -m pytest tests/integration/test_media_hardening.py tests/integration/test_exif_checkpoints.py -q
-```
-
-## Concurrency and crash recovery (US07-04)
-
-Crash safety is proven by crashing. `photo_pipeline/faults.py` defines the control
-points — the persisted transitions of the rename, archive, EXIF, upload, and job
-lanes — and arms one only when `PHOTO_PIPELINE_FAULT_AFTER` names it, at which
-point the process dies the way a `SIGKILL` does. There is no endpoint and no
-configuration field that can reach a barrier; a deployment that never sets the
-variable can never hit one.
-
-The race suite runs each scenario several times with a seed recorded on the test
-result (`race_seed`) and asserts invariants rather than schedules: work is never
-claimed or executed twice, a stale fencing token never commits, no file body is
-lost or overwritten, and the database still passes `PRAGMA integrity_check`.
-
-```bash
-work_item/scripts/python -m pytest tests/integration/test_concurrency_races.py \
-    tests/integration/test_fault_matrix.py tests/e2e/test_crash_recovery.py -q
-
-# replay a failure, or soak for new interleavings
-PHOTO_PIPELINE_RACE_SEED=1234 PHOTO_PIPELINE_RACE_REPEATS=50 \
-    work_item/scripts/python -m pytest tests/integration/test_concurrency_races.py -q
-```
-
-Any failing test keeps its evidence: the temporary database (with its write-ahead
-log), the journals, the logs, the recorded seed, and a SHA-256 manifest of every
-file in the temporary library are copied to `.artifacts/<test id>/` before pytest
-deletes the directory. Point `PHOTO_PIPELINE_TEST_ARTIFACTS` elsewhere to collect
-them from CI.
-
 ## Legacy CLI archive
 
 The command-line tools this application was extracted from are frozen in

legacy_cli_archive/donor_ledger.yaml

@@ -480,13 +480,6 @@ rows:
       inference path needs the local model + deterministic fake.
     target: photo_pipeline/integrations/nsfw_model.py
     tests: [test_nsfwtag::test_score_images_cache_hit_skips_model]
-    delta: >
-      The donor set Pillow's process-global ImageFile.LOAD_TRUNCATED_IMAGES so a
-      partially downloaded file still scored. Here the same process also hashes
-      files and renders previews, and those must keep refusing a truncated file
-      rather than silently working on half of one; scoring opens images through
-      the bounded photo_pipeline.imaging door instead and skips the ones it cannot
-      read, leaving them unscored and visibly undecided (US07-03).
     status: characterized
 
   - id: nt-exif-keyword

migrations/versions/0015_exif_projections.py

@@ -1,42 +0,0 @@
-"""Durable EXIF projections per asset and stage (US07-03).
-
-Revision ID: 0015_exif_projections
-Revises: 0014_restore_plans
-Create Date: 2026-08-16
-
-The concept's ``exif_projections`` table, added at the point it earns its keep: a
-checkpoint that finds a field it does not own changed must be able to say so after
-a restart. ``state`` is verified | divergent | failed, and only ``verified`` counts
-as a completed metadata stage.
-"""
-
-import sqlalchemy as sa
-from alembic import op
-
-revision = "0015_exif_projections"
-down_revision = "0014_restore_plans"
-branch_labels = None
-depends_on = None
-
-
-def upgrade() -> None:
-    op.create_table(
-        "exif_projections",
-        sa.Column("asset_id", sa.String(), sa.ForeignKey("assets.id"), primary_key=True),
-        sa.Column("stage", sa.String(), primary_key=True),  # safety | analysis
-        sa.Column("id", sa.String(), nullable=False),
-        sa.Column("projection_version", sa.Integer(), nullable=False, server_default="0"),
-        # What the stage asked for: {"add": [...], "remove": [...]}.
-        sa.Column("desired_json", sa.String(), nullable=True),
-        # Fields outside the stage's ownership that did not survive the write.
-        sa.Column("divergent_fields", sa.String(), nullable=True),
-        sa.Column("result_file_sha256", sa.String(), nullable=True),
-        sa.Column("state", sa.String(), nullable=False),
-        sa.Column("error_code", sa.String(), nullable=True),
-        sa.Column("verified_at", sa.DateTime(timezone=True), nullable=True),
-        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=True),
-    )
-
-
-def downgrade() -> None:
-    op.drop_table("exif_projections")

photo_pipeline/api/app.py

@@ -41,7 +41,6 @@ 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
-from photo_pipeline.services.thumbnails import ThumbnailService
 from photo_pipeline.services.upload_batches import UploadBatchService
 
 FRONTEND_DIR = Path(__file__).resolve().parents[2] / "frontend"
@@ -97,9 +96,6 @@ def create_app(config: Config | None = None) -> FastAPI:
         # An upload whose process died left no outcome behind; resolve it now so the
         # uploader lane is free and the uncertain batch is visible (US05-02).
         UploadBatchService(app.state.session_factory, config=config).recover()
-        # A render killed mid-write leaves its temporary beside the cache entry;
-        # remove those recognized leftovers, and only those (US07-03).
-        ThumbnailService(app.state.session_factory, config).cleanup_temp_files()
         try:
             yield
         finally:

photo_pipeline/faults.py

@@ -1,45 +0,0 @@
-"""Test-only fault control points (concept §18, US07-04).
-
-Crash safety can only be proven by crashing at the exact moment a transition has
-been persisted but its consequence has not. That needs a barrier *inside* the
-production code path — but not a production capability: there is no endpoint, no
-service method, and no configuration file entry that can trigger one. The only
-switch is an environment variable naming a single point, read at the moment it is
-passed, and the only thing it does is kill the process. A deployment that never
-sets it can never reach the barrier.
-
-``os._exit`` is deliberate: it skips atexit handlers, buffered flushes, and
-``finally`` blocks, which is what a real ``SIGKILL`` or power loss does. A clean
-shutdown would prove nothing.
-
-The points are the persisted transitions of the journalled stages:
-
-    rename    moving | moved | database_updated | verified | complete
-    archive   transferring | verified | removing | source_removed | complete
-    exif      exif:written        — keywords on disk, checkpoint not yet recorded
-    upload    upload:accepted     — uploader exited, outcome not yet persisted
-    jobs      job:item_done       — item committed, job outcome not yet written
-
-Recovery for each is asserted in tests/integration/test_fault_matrix.py and
-tests/e2e/test_crash_recovery.py.
-"""
-
-from __future__ import annotations
-
-import os
-
-ENV_VAR = "PHOTO_PIPELINE_FAULT_AFTER"
-
-EXIF_WRITTEN = "exif:written"
-UPLOAD_ACCEPTED = "upload:accepted"
-JOB_ITEM_DONE = "job:item_done"
-
-
-def maybe_fault(point: str) -> None:
-    """Die abruptly when ``PHOTO_PIPELINE_FAULT_AFTER`` names ``point``.
-
-    Shared by the rename, archive, restore, EXIF, upload, and job lanes, each
-    passing its own state names. Never set the variable outside tests.
-    """
-    if os.environ.get(ENV_VAR) == point:
-        os._exit(9)

photo_pipeline/imaging.py

@@ -1,119 +0,0 @@
-"""Bounded, defensive image decoding — one door for every pixel this app reads.
-
-A photo library contains files nobody planned for: truncated downloads, zero-byte
-placeholders, a PNG whose header claims 200000×200000, a TIFF with a broken ICC
-profile, an extension that lies about its content. None of them may take down a
-request or a worker, and none may decode more pixels than the caller allowed
-(concept §17: decoded pixels, not file size, are what exhausts memory).
-
-``open_image`` is that single door:
-
-* the declared dimensions are checked **before** a pixel is decoded;
-* Pillow's decompression-bomb *warning* is promoted to an error, because the
-  warning band (between Pillow's limit and twice it) still decodes the image;
-* every decoder failure — at open time or during the caller's decode — becomes one
-  of two typed errors, so callers map them to their own item state instead of
-  catching ``Exception``;
-* error text names no path: it reaches API responses, and the full reason goes to
-  the server log instead (US07-02).
-
-``to_srgb`` and ``draft`` are the other two bounded-decode helpers: colour-manage a
-profile-bearing image into sRGB, and let JPEG decode straight to a size near the
-requested one rather than at full resolution.
-"""
-
-from __future__ import annotations
-
-import io
-import logging
-import warnings
-from contextlib import contextmanager
-from pathlib import Path
-
-from PIL import Image, ImageCms, ImageFile, UnidentifiedImageError
-
-log = logging.getLogger(__name__)
-
-# Matches ``Config.thumbnail_max_pixels``; used where no configuration is at hand
-# (hashing runs inside discovery, which takes no config).
-DEFAULT_MAX_PIXELS = 100_000_000
-
-
-class MediaError(Exception):
-    """A file could not be turned into pixels safely."""
-
-
-class UndecodableImage(MediaError):
-    """Corrupt, truncated, empty, or not an image at all."""
-
-
-class ImageTooLarge(MediaError):
-    """More pixels than this operation is allowed to decode."""
-
-
-@contextmanager
-def open_image(path: Path | str, *, max_pixels: int = DEFAULT_MAX_PIXELS):
-    """Yield an open :class:`PIL.Image.Image`, bounded and with typed failures.
-
-    Decoder errors raised inside the ``with`` body are translated too — a truncated
-    JPEG only fails when its pixels are actually pulled, which is the caller's line,
-    not this one.
-    """
-    # Pillow's truncation tolerance is a process-global switch that any library in
-    # the process can flip (the donor CLI did). This door decides the policy for its
-    # own callers: half a file is not a picture.
-    tolerated = ImageFile.LOAD_TRUNCATED_IMAGES
-    ImageFile.LOAD_TRUNCATED_IMAGES = False
-    with warnings.catch_warnings():
-        # The warning band is not a warning for us: it means Pillow was willing to
-        # decode an image large enough to be a denial-of-service.
-        warnings.simplefilter("error", Image.DecompressionBombWarning)
-        try:
-            with Image.open(path) as image:
-                width, height = image.size
-                if width * height > max_pixels:
-                    raise ImageTooLarge(f"{width}x{height} exceeds the {max_pixels} pixel limit")
-                yield image
-        except MediaError:
-            raise
-        except (Image.DecompressionBombError, Image.DecompressionBombWarning) as error:
-            log.info("refused oversized image %s: %s", path, error)
-            raise ImageTooLarge("image exceeds the decompression-bomb limit") from error
-        except (UnidentifiedImageError, OSError, ValueError, SyntaxError, MemoryError) as error:
-            log.info("cannot decode %s: %s", path, error)
-            raise UndecodableImage(f"cannot decode image ({type(error).__name__})") from error
-        finally:
-            ImageFile.LOAD_TRUNCATED_IMAGES = tolerated
-
-
-def draft(image: Image.Image, size: int) -> None:
-    """Ask the decoder for a smaller image where the format allows it (JPEG).
-
-    This is the difference between decoding a 40-megapixel JPEG and decoding the
-    roughly 1-megapixel version a 1280px preview needs.
-    """
-    try:
-        image.draft(None, (size, size))
-    except (AttributeError, ValueError, OSError):  # not a draft-capable format
-        pass
-
-
-def to_srgb(image: Image.Image, *, mode: str) -> Image.Image:
-    """Convert into ``mode``, colour-managing through an embedded ICC profile.
-
-    Without this a wide-gamut original renders with visibly wrong colours, because
-    its numbers are interpreted as sRGB. A broken or unreadable profile is not a
-    reason to fail a preview — the plain conversion is still a correct picture.
-    """
-    profile = image.info.get("icc_profile")
-    if profile:
-        try:
-            return ImageCms.profileToProfile(
-                image,
-                ImageCms.ImageCmsProfile(io.BytesIO(profile)),
-                ImageCms.createProfile("sRGB"),
-                outputMode=mode,
-            )
-        except Exception as error:  # noqa: BLE001 - any ICC failure falls back
-            log.info("ignoring unusable ICC profile on %s: %s", getattr(image, "filename", "?"), error)
-    return image.convert(mode)

photo_pipeline/integrations/exiftool.py

@@ -17,20 +17,6 @@ import os
 import subprocess
 from collections.abc import Iterable
 
-# A hung exiftool must not hang the worker with it: every call is bounded, and a
-# call that runs out of time is treated exactly like a failed one — no metadata
-# answer, nothing marked verified (US07-04). The knob exists because "slow" is a
-# property of the machine, not of the code: huge files on a slow network volume
-# legitimately take longer than the default.
-DEFAULT_TIMEOUT_SECONDS = 120.0
-
-
-def _timeout() -> float:
-    try:
-        return float(os.environ.get("PHOTO_PIPELINE_EXIFTOOL_TIMEOUT", DEFAULT_TIMEOUT_SECONDS))
-    except ValueError:
-        return DEFAULT_TIMEOUT_SECONDS
-
 
 def read_keyword_sets(paths: Iterable[str]) -> dict[str, set[str]]:
     """Map each path to its lowercased set of ``Keywords`` + ``Subject`` values.
@@ -48,9 +34,8 @@ def read_keyword_sets(paths: Iterable[str]) -> dict[str, set[str]]:
             input="\n".join(paths),
             capture_output=True,
             text=True,
-            timeout=_timeout(),
         )
-    except (FileNotFoundError, subprocess.TimeoutExpired):
+    except FileNotFoundError:
         return {}
     out: dict[str, set[str]] = {}
     try:
@@ -71,33 +56,6 @@ def read_keyword_sets(paths: Iterable[str]) -> dict[str, set[str]]:
     return out
 
 
-def read_all(path: str) -> dict | None:
-    """Every tag exiftool can read from ``path``, or ``None`` when it cannot answer.
-
-    This is the snapshot an EXIF checkpoint compares against: proving that a write
-    preserved the fields it does not own requires knowing all of them, not just the
-    ones being written (US07-03). ``None`` (exiftool missing, unreadable file,
-    unparsable output) is not an empty snapshot — a caller must not read it as
-    "nothing was there".
-    """
-    try:
-        result = subprocess.run(
-            ["exiftool", "-m", "-j", "-G0:1", path],
-            capture_output=True,
-            text=True,
-            timeout=_timeout(),
-        )
-    except (FileNotFoundError, subprocess.TimeoutExpired):
-        return None
-    try:
-        records = json.loads(result.stdout or "[]")
-    except ValueError:
-        return None
-    if not records:
-        return None
-    return {k: v for k, v in records[0].items() if k != "SourceFile"}
-
-
 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"]
@@ -109,10 +67,4 @@ def apply_keywords(path: str, *, add: Iterable[str] = (), remove: Iterable[str]
     if len(args) == 3:
         return True
     args.append(path)
-    try:
-        return subprocess.run(
-            args, capture_output=True, text=True, timeout=_timeout()
-        ).returncode == 0
-    except (FileNotFoundError, subprocess.TimeoutExpired):
-        # A write that never returned is not a write that succeeded.
-        return False
+    return subprocess.run(args, capture_output=True, text=True).returncode == 0

photo_pipeline/integrations/nsfw_model.py

@@ -15,8 +15,6 @@ from __future__ import annotations
 
 from pathlib import Path
 
-from photo_pipeline import imaging
-
 MODEL_ID = "AdamCodd/vit-base-nsfw-detector"
 BATCH = 16
 
@@ -56,17 +54,13 @@ class NsfwModel:
         self._ensure_loaded()
         import numpy as np
         import torch
-        from PIL import Image
+        from PIL import Image, ImageFile
 
-        # The donor set ``ImageFile.LOAD_TRUNCATED_IMAGES = True`` here. That flag is
-        # process-global: in this application the same process also hashes files and
-        # renders previews, and those must keep failing loudly on a truncated file
-        # rather than quietly working on half of one (US07-03). An unreadable image
-        # is skipped instead — it stays unscored, and therefore visibly undecided.
-        def preprocess(path):
-            with imaging.open_image(path) as image:
-                small = image.convert("RGB").resize((self._size, self._size), Image.BILINEAR)
-            array = (np.asarray(small, dtype="float32") / 255.0 - 0.5) / 0.5
+        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]] = []
@@ -75,10 +69,9 @@ class NsfwModel:
             tensors, batch_paths = [], []
             for path in items[start : start + self.batch]:
                 try:
-                    tensors.append(preprocess(path))
+                    tensors.append(preprocess(Image.open(path)))
                     batch_paths.append(path)
-                except (imaging.MediaError, OSError, ValueError):
-                    # One bad file must not cost the batch its other fifteen.
+                except Exception:
                     continue
             if not tensors:
                 continue

photo_pipeline/jobs/worker.py

@@ -16,7 +16,6 @@ from collections.abc import Mapping, Sequence
 from sqlalchemy import select
 from sqlalchemy.orm import sessionmaker
 
-from photo_pipeline.faults import JOB_ITEM_DONE, maybe_fault
 from photo_pipeline.jobs.handlers import REGISTRY, Cancelled, Handler, JobContext
 from photo_pipeline.models import JobItem
 from photo_pipeline.services.jobs import ItemState, JobConflict, JobService, JobState
@@ -87,9 +86,6 @@ class Worker:
                 )
             else:
                 self.service.set_item(job_id, item_key, ItemState.SUCCEEDED, fencing_token=token)
-                # One item is durably done, the job outcome is not: the control point
-                # for a crash mid-batch (US07-04). Recovery must not re-run this item.
-                maybe_fault(JOB_ITEM_DONE)
             self.service.heartbeat(job_id, self.worker_id, lease_seconds=self.lease_seconds)
 
         self._finalize(job_id, token, cancelled=cancelled, any_failed=any_failed)

photo_pipeline/models/__init__.py

@@ -7,7 +7,6 @@ Alembic environment relies on.
 from photo_pipeline.models.albums import AlbumProposal
 from photo_pipeline.models.archives import ArchiveLocation, ArchiveOperation, ArchivePlan
 from photo_pipeline.models.assets import Asset, AssetPath
-from photo_pipeline.models.exif import ExifProjection
 from photo_pipeline.models.duplicates import (
     DuplicateCluster,
     DuplicateMember,
@@ -29,7 +28,6 @@ __all__ = [
     "DuplicateCluster",
     "DuplicateMember",
     "DuplicateNegativeLink",
-    "ExifProjection",
     "Job",
     "JobItem",
     "JobEvent",

photo_pipeline/models/exif.py

@@ -1,33 +0,0 @@
-"""The durable EXIF projection per asset and stage (concept §3, US07-03).
-
-One row per ``(asset_id, stage)``: what the stage wanted written, what the file
-looked like afterwards, and whether anything outside the stage's ownership moved.
-``state = divergent`` is the whole point of the table — it survives restarts, keeps
-the asset out of stages that require verified metadata, and gives a human something
-to look at instead of a silent repair.
-"""
-
-from __future__ import annotations
-
-from datetime import datetime
-
-from sqlalchemy import DateTime, ForeignKey, Integer, String
-from sqlalchemy.orm import Mapped, mapped_column
-
-from photo_pipeline.db import Base
-
-
-class ExifProjection(Base):
-    __tablename__ = "exif_projections"
-
-    asset_id: Mapped[str] = mapped_column(ForeignKey("assets.id"), primary_key=True)
-    stage: Mapped[str] = mapped_column(String, primary_key=True)  # safety | analysis
-    id: Mapped[str] = mapped_column(String, nullable=False)
-    projection_version: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
-    desired_json: Mapped[str | None] = mapped_column(String)
-    divergent_fields: Mapped[str | None] = mapped_column(String)  # JSON array
-    result_file_sha256: Mapped[str | None] = mapped_column(String)
-    state: Mapped[str] = mapped_column(String, nullable=False)  # verified|divergent|failed
-    error_code: Mapped[str | None] = mapped_column(String)
-    verified_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
-    updated_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))

photo_pipeline/services/analysis.py

@@ -26,8 +26,9 @@ from sqlalchemy import func, select
 from sqlalchemy.orm import sessionmaker
 
 from photo_pipeline import path_policy
+from photo_pipeline.integrations import exiftool
 from photo_pipeline.models import AnalysisResult, Asset, SafetyReview
-from photo_pipeline.services import exif_checkpoint
+from photo_pipeline.services import hashing
 from photo_pipeline.services.safety import SFW
 
 MODEL = "gemini-2.5-flash"
@@ -78,11 +79,6 @@ class AnalysisService:
             latest[review.asset_id] = review.decision
         return {aid for aid, decision in latest.items() if decision == SFW}
 
-    def _is_still_sfw(self, asset_id: str) -> bool:
-        """Re-read the current safety decision straight from the database."""
-        with self._session_factory() as session:
-            return asset_id in self._sfw_asset_ids(session)
-
     def eligible_asset_ids(self) -> list[str]:
         """Confirmed-SFW canonical active assets without a completed analysis."""
         with self._session_factory() as session:
@@ -166,22 +162,6 @@ class AnalysisService:
                 self._store(asset_id, status="error", result=None, error=str(error), tokens=0, raw="")
                 errors += 1
                 continue
-            # Third gate, after the call: a provider request takes seconds, and the
-            # reviewer may have flipped this asset to NSFW while it was in flight.
-            # The result describes an asset that is no longer analysable, so it is
-            # discarded — not stored, and above all not written into its EXIF
-            # (concept §18 scenario 7, US07-04).
-            if not self._is_still_sfw(asset_id):
-                self._store(
-                    asset_id,
-                    status="skipped_nsfw",
-                    result=None,
-                    error="the safety decision changed while analysis was in flight",
-                    tokens=0,
-                    raw="",
-                )
-                skipped += 1
-                continue
             self._store(
                 asset_id,
                 status="analyzed",
@@ -190,7 +170,7 @@ class AnalysisService:
                 tokens=int(result.get("_tokens", 0)) if isinstance(result, dict) else 0,
                 raw=json.dumps(result, ensure_ascii=False),
             )
-            self._write_analysis_exif(asset_id, path, result)
+            _write_analysis_exif(path, result)
             analyzed += 1
         return {"analyzed": analyzed, "skipped": skipped, "errors": errors}
 
@@ -215,42 +195,10 @@ class AnalysisService:
                 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 _write_analysis_exif(self, asset_id: str, path: str, result: dict) -> None:
-        """The analysis EXIF checkpoint: additive keywords, then prove the rest held.
-
-        Additive by design — safety keywords and the user's own keywords are merged
-        with, never replaced (concept §3). ``exif_written_at`` is set only when the
-        read-back verified both the new keywords and every field this stage does not
-        own; a divergent result is recorded and left for a human (US07-03).
-        ponytail: the managed ``AI:`` caption segment, once captions are owned here.
-        """
-        tags = tuple(str(tag) for tag in (result.get("tags") or []))
-        if not tags:
-            return
-        checkpoint = exif_checkpoint.run(path, add=tags)
-        exif_checkpoint.record(
-            self._session_factory,
-            asset_id=asset_id,
-            stage="analysis",
-            result=checkpoint,
-            add=tags,
-        )
-        if not checkpoint.verified:
-            return
-        with self._session_factory() as session:
-            row = session.get(AnalysisResult, asset_id)
-            if row is not None:
-                row.exif_written_at = checkpoint.verified_at
-            asset = session.get(Asset, asset_id)
-            if asset is not None and checkpoint.sha256:
-                # The bytes changed when the container was rewritten; upload must use
-                # the hash of what is actually on disk now (concept §3).
-                asset.current_sha256 = checkpoint.sha256
-            session.commit()
-
     def get(self, asset_id: str) -> dict | None:
         with self._session_factory() as session:
             row = session.get(AnalysisResult, asset_id)
@@ -263,6 +211,17 @@ def _album_hint(path: str) -> str:
     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 []

photo_pipeline/services/archive_transfer.py

@@ -52,7 +52,6 @@ from sqlalchemy import select
 from sqlalchemy.orm import sessionmaker
 
 from photo_pipeline.config import Config
-from photo_pipeline.faults import maybe_fault
 from photo_pipeline.models import ArchiveLocation, ArchiveOperation, ArchivePlan, Asset, AssetPath
 from photo_pipeline.services.archive_journal import (
     ARCHIVE,
@@ -64,7 +63,7 @@ from photo_pipeline.services.archive_journal import (
 from photo_pipeline.services.archives import MARKER_NAME, ArchiveError, ArchiveService
 from photo_pipeline.services.duplicates import DuplicateService
 from photo_pipeline.services.hashing import sha256_file
-from photo_pipeline.services.rename_apply import PreconditionFailed
+from photo_pipeline.services.rename_apply import PreconditionFailed, maybe_fault
 from photo_pipeline.services.thumbnails import ThumbnailService
 
 # The per-medium manifest: one JSON line per archived file, appended and fsynced

photo_pipeline/services/exif_checkpoint.py

@@ -1,178 +0,0 @@
-"""The EXIF checkpoint every metadata stage ends with (concept §3, US07-03).
-
-A stage does not own a file's metadata; it owns a few fields in it. So writing is
-never "set these tags" — it is:
-
-    snapshot everything → write only the owned fields → read everything back
-    → prove the owned fields landed → prove nothing else moved
-    → refresh the file hash → record the projection
-
-Non-destructive here means *semantic* preservation: exiftool may rewrite the whole
-container, so the file's bytes, size, and timestamps legitimately change. What may
-not change is any field this stage does not own. When one does, the checkpoint is
-``divergent``: the result is recorded, the stage is **not** marked verified, and
-nothing is silently repaired — a later stage that needs verified metadata (upload)
-therefore stays blocked until a human looks.
-
-``failed`` is the third outcome and is deliberately distinct: exiftool missing, an
-unreadable file, or a write that did not take is not evidence that metadata is fine.
-"""
-
-from __future__ import annotations
-
-import json
-import uuid
-from dataclasses import dataclass
-from datetime import datetime, timezone
-
-from photo_pipeline.faults import EXIF_WRITTEN, maybe_fault
-from photo_pipeline.integrations import exiftool
-from photo_pipeline.models import ExifProjection
-from photo_pipeline.services import hashing
-
-# The tags the safety and analysis stages may change. Matched on the tag name, so
-# every group carries the same rule (IPTC:Keywords, XMP:XMP-dc:Subject, ...).
-OWNED_TAGS = frozenset({"Keywords", "Subject"})
-
-# Not metadata about the picture: filesystem facts, the digest that necessarily
-# moves whenever IPTC does, and the structural tags exiftool has to create the first
-# time it writes an IPTC or XMP block. Comparing these would report every write as
-# divergent and make the signal worthless.
-VOLATILE_PREFIXES = ("File:System:", "ExifTool:")
-VOLATILE_KEYS = frozenset(
-    {
-        "File:CurrentIPTCDigest",
-        "IPTC:ApplicationRecordVersion",
-        "XMP:XMP-x:XMPToolkit",
-        "XMP:XMP-xmp:MetadataDate",
-    }
-)
-
-VERIFIED = "verified"
-DIVERGENT = "divergent"
-FAILED = "failed"
-
-
-@dataclass(frozen=True)
-class CheckpointResult:
-    state: str  # verified | divergent | failed
-    changed_fields: tuple[str, ...] = ()
-    sha256: str | None = None
-    verified_at: datetime | None = None
-    reason: str | None = None
-
-    @property
-    def verified(self) -> bool:
-        return self.state == VERIFIED
-
-
-def _now() -> datetime:
-    return datetime.now(timezone.utc)
-
-
-def is_owned(key: str) -> bool:
-    return key.rsplit(":", 1)[-1] in OWNED_TAGS
-
-
-def is_volatile(key: str) -> bool:
-    return key in VOLATILE_KEYS or key.startswith(VOLATILE_PREFIXES)
-
-
-def compare(before: dict, after: dict) -> tuple[str, ...]:
-    """Fields outside this stage's ownership whose value did not survive the write.
-
-    Additions count: a tag that appears out of nowhere is as much a divergence as a
-    tag that disappeared — both mean the write did more than it was asked to.
-    """
-    keys = set(before) | set(after)
-    return tuple(
-        sorted(
-            key
-            for key in keys
-            if not is_owned(key)
-            and not is_volatile(key)
-            and before.get(key) != after.get(key)
-        )
-    )
-
-
-def owned_values(snapshot: dict) -> set[str]:
-    """Lowercased Keywords/Subject values across every group in the snapshot."""
-    values: set[str] = set()
-    for key, value in snapshot.items():
-        if not is_owned(key):
-            continue
-        items = value if isinstance(value, list) else [value]
-        values.update(str(item).strip().lower() for item in items if item is not None)
-    return values
-
-
-def run(
-    path: str,
-    *,
-    add: tuple[str, ...] = (),
-    remove: tuple[str, ...] = (),
-) -> CheckpointResult:
-    """Write the owned keywords for one asset and verify the whole file around them."""
-    before = exiftool.read_all(path)
-    if before is None:
-        return CheckpointResult(FAILED, reason="metadata_unreadable")
-
-    if not exiftool.apply_keywords(path, add=add, remove=remove):
-        return CheckpointResult(FAILED, reason="write_failed")
-
-    # The file on disk has changed; nothing about it is recorded yet. A crash here
-    # is the worst case for metadata, so it is a fault control point (US07-04).
-    maybe_fault(EXIF_WRITTEN)
-
-    after = exiftool.read_all(path)
-    if after is None:
-        return CheckpointResult(FAILED, reason="readback_unreadable")
-
-    present = owned_values(after)
-    wanted = {value.strip().lower() for value in add}
-    unwanted = {value.strip().lower() for value in remove}
-    if not wanted <= present or (unwanted & present):
-        return CheckpointResult(FAILED, reason="owned_fields_not_written")
-
-    changed = compare(before, after)
-    sha256 = hashing.sha256_file(path)
-    if changed:
-        return CheckpointResult(DIVERGENT, changed_fields=changed, sha256=sha256)
-    return CheckpointResult(VERIFIED, sha256=sha256, verified_at=_now())
-
-
-def record(
-    session_factory,
-    *,
-    asset_id: str,
-    stage: str,
-    result: CheckpointResult,
-    add: tuple[str, ...] = (),
-    remove: tuple[str, ...] = (),
-) -> None:
-    """Persist the projection for ``(asset_id, stage)`` — one current row per pair.
-
-    The row is what makes divergence durable and reviewable rather than a log line
-    that scrolled away.
-    """
-    with session_factory() as session:
-        row = session.get(ExifProjection, (asset_id, stage))
-        if row is None:
-            row = ExifProjection(asset_id=asset_id, stage=stage, id=str(uuid.uuid4()))
-            session.add(row)
-        row.projection_version = (row.projection_version or 0) + 1
-        row.desired_json = json.dumps({"add": list(add), "remove": list(remove)})
-        row.divergent_fields = json.dumps(list(result.changed_fields))
-        row.result_file_sha256 = result.sha256
-        row.state = result.state
-        row.error_code = result.reason
-        row.verified_at = result.verified_at
-        row.updated_at = _now()
-        session.commit()
-
-
-def state_for(session_factory, asset_id: str, stage: str) -> str | None:
-    with session_factory() as session:
-        row = session.get(ExifProjection, (asset_id, stage))
-        return row.state if row else None

photo_pipeline/services/hashing.py

@@ -16,9 +16,7 @@ from __future__ import annotations
 import hashlib
 from pathlib import Path
 
-from PIL import ImageOps
-
-from photo_pipeline import imaging
+from PIL import Image, ImageOps
 
 PIXEL_HASH_VERSION = 1
 PHASH_VERSION = 1
@@ -43,9 +41,7 @@ def _digest_file(path: Path | str, digest) -> str:
 
 
 def pixel_sha256(path: Path | str) -> str:
-    # Bounded decode: a header claiming a billion pixels is refused before it is
-    # ever turned into memory (US07-03).
-    with imaging.open_image(path) as image:
+    with Image.open(path) as image:
         oriented = ImageOps.exif_transpose(image)
         rgb = oriented.convert("RGB")
         header = f"{PIXEL_HASH_VERSION}:{rgb.width}x{rgb.height}:".encode()
@@ -72,9 +68,7 @@ def phash(path: Path | str) -> str:
     import numpy as np
     from scipy.fftpack import dct
 
-    from PIL import Image
-
-    with imaging.open_image(path) as image:
+    with Image.open(path) as image:
         small = image.convert("L").resize((32, 32), Image.LANCZOS)
         matrix = np.asarray(small, dtype=np.float64)
     transformed = dct(dct(matrix, axis=0), axis=1)

photo_pipeline/services/jobs.py

@@ -236,32 +236,17 @@ class JobService:
                 raise InvalidTransition(f"{job.state} -> {to_state}")
             if worker_id is not None and job.lease_owner not in (None, worker_id):
                 raise JobConflict(f"job {job_id} owned by {job.lease_owner}, not {worker_id}")
-
-            # Compare-and-set on the version this decision was made against. Without
-            # it a transition validated against a row that has since been claimed,
-            # cancelled, or finished would overwrite that newer state (concept §16
-            # database rule 6) — a cancel racing a claim used to un-claim a running
-            # job and leave the worker finalizing a job it no longer owned.
-            values = {
-                "state": to_state,
-                "version": job.version + 1,
-                "updated_at": now,
-            }
+            job.state = to_state
+            job.version += 1
+            job.updated_at = now
             if error:
-                values["error_code"], values["error_message"] = error
+                job.error_code, job.error_message = error
             if to_state in TERMINAL_STATES:
-                values.update(finished_at=now, lease_owner=None, lease_expires_at=None)
-            result = session.execute(
-                update(Job).where(Job.id == job_id, Job.version == job.version).values(**values)
-            )
-            if result.rowcount != 1:
-                session.rollback()
-                raise JobConflict(
-                    f"job {job_id} changed while transitioning to {to_state}; retry"
-                )
+                job.finished_at = now
+                job.lease_owner = None
+                job.lease_expires_at = None
             self._event(session, job_id, f"state:{to_state}", error[1] if error else None)
             session.commit()
-            session.expire_all()  # the core UPDATE bypassed the identity map
             return self._snapshot(session, job_id)
 
     def cancel(self, job_id: str) -> dict:

photo_pipeline/services/rename_apply.py

@@ -46,11 +46,9 @@ from pathlib import Path
 from sqlalchemy import select
 from sqlalchemy.orm import sessionmaker
 
-from photo_pipeline.faults import maybe_fault
 from photo_pipeline.models import Asset, AssetPath, RenamePlan
 from photo_pipeline.services import hashing
 from photo_pipeline.services.rename_journal import (
-    ALLOWED_TRANSITIONS,
     MANUAL,
     RESUMABLE,
     JournalState,
@@ -85,6 +83,18 @@ def _now() -> datetime:
     return datetime.now(timezone.utc)
 
 
+def maybe_fault(state: str) -> None:
+    """Test-only crash barrier (concept §18 fault injection).
+
+    When ``PHOTO_PIPELINE_FAULT_AFTER`` names a journal state, the process dies
+    abruptly the moment that state has been persisted — modelling a real kill at
+    exactly that transition. Never set outside tests. Shared with the archive
+    transfer journal (US06-02), which uses the same env var and its own state names.
+    """
+    if os.environ.get("PHOTO_PIPELINE_FAULT_AFTER") == state:
+        os._exit(9)
+
+
 class RenameApplyService:
     def __init__(self, session_factory: sessionmaker, *, library_roots: tuple = ()) -> None:
         self._session_factory = session_factory
@@ -133,10 +143,20 @@ class RenameApplyService:
                 self._apply_one(operation, token=token, worker_id=worker_id)
                 applied += 1
             except PreconditionFailed as error:
-                self._record_failure(operation["id"], token, error.code, str(error))
+                self.journal.transition(
+                    operation["id"],
+                    JournalState.FAILED,
+                    fencing_token=token,
+                    error=(error.code, str(error)),
+                )
                 failed += 1
             except Exception as error:  # unexpected: record and stop touching disk
-                self._record_failure(operation["id"], token, "apply_error", str(error))
+                self.journal.transition(
+                    operation["id"],
+                    JournalState.FAILED,
+                    fencing_token=token,
+                    error=("apply_error", str(error)),
+                )
                 failed += 1
         state = self.journal.sync_plan_state(plan_id)
         return {
@@ -147,26 +167,6 @@ class RenameApplyService:
             "state": state,
         }
 
-    def _record_failure(self, operation_id: str, token: int, code: str, message: str) -> None:
-        """Record a failed operation in a state its journal can actually reach.
-
-        ``failed`` only makes sense while nothing has moved. Once the folder is at
-        its destination — a postcondition failure such as bytes edited during the
-        move — the operation is not "failed and forgotten": the disk changed and
-        the database followed, so it becomes ``rollback_required`` and waits for a
-        human (US07-04). Guessing an unreachable transition used to raise out of
-        ``apply`` and lose the record entirely.
-        """
-        current = self.journal.get(operation_id)["journal_state"]
-        target = (
-            JournalState.FAILED
-            if JournalState.FAILED in ALLOWED_TRANSITIONS.get(current, set())
-            else JournalState.ROLLBACK_REQUIRED
-        )
-        self.journal.transition(
-            operation_id, target, fencing_token=token, error=(code, message)
-        )
-
     def _apply_one(self, operation: dict, *, token: int, worker_id: str) -> None:
         source = Path(operation["source_path"])
         destination = Path(operation["destination_path"])

photo_pipeline/services/rename_journal.py

@@ -81,17 +81,7 @@ ALLOWED_TRANSITIONS = {
 
 TERMINAL_STATES = frozenset({JournalState.COMPLETE, JournalState.ROLLED_BACK})
 # States where the disk may already have been touched by this operation.
-# ``rollback_required`` belongs here too (US07-04): the move happened and someone
-# has to decide what to do about it, so the library is not in a state another
-# mutation may build on.
-UNSAFE_STATES = frozenset(
-    {
-        JournalState.MOVING,
-        JournalState.MOVED,
-        JournalState.DATABASE_UPDATED,
-        JournalState.ROLLBACK_REQUIRED,
-    }
-)
+UNSAFE_STATES = frozenset({JournalState.MOVING, JournalState.MOVED, JournalState.DATABASE_UPDATED})
 
 RESUMABLE = "resumable"
 ROLLBACK_SAFE = "rollback_safe"

photo_pipeline/services/restores.py

@@ -53,7 +53,6 @@ from sqlalchemy import select
 from sqlalchemy.orm import sessionmaker
 
 from photo_pipeline.config import Config
-from photo_pipeline.faults import maybe_fault
 from photo_pipeline.jobs.domain_handlers import ARCHIVE_LOCK, LIBRARY_WRITE_LOCK, UPLOAD_LOCK
 from photo_pipeline.models import ArchiveLocation, ArchiveOperation, ArchivePlan, Asset, AssetPath
 from photo_pipeline.path_policy import PathPolicyError, is_excluded, normalize_root, resolve_within
@@ -74,7 +73,7 @@ from photo_pipeline.services.archive_transfer import (
 from photo_pipeline.services.archives import ArchiveError
 from photo_pipeline.services.hashing import sha256_file
 from photo_pipeline.services.jobs import JobService
-from photo_pipeline.services.rename_apply import PreconditionFailed
+from photo_pipeline.services.rename_apply import PreconditionFailed, maybe_fault
 from photo_pipeline.services.rename_journal import RenameJournal
 
 PREFLIGHT_VERSION = 1

photo_pipeline/services/safety.py

@@ -108,8 +108,9 @@ from datetime import datetime, timezone
 from sqlalchemy import select
 from sqlalchemy.orm import sessionmaker
 
-from photo_pipeline.models import Asset, ExifProjection, SafetyReview
-from photo_pipeline.services import exif_checkpoint
+from photo_pipeline.integrations import exiftool
+from photo_pipeline.models import Asset, SafetyReview
+from photo_pipeline.services import hashing
 
 DECISIONS = {SFW, NSFW, "deferred"}
 
@@ -173,14 +174,6 @@ class SafetyService:
         with self._session_factory() as session:
             assets = list(session.scalars(_eligible_assets_query().order_by(Asset.current_path)))
             latest = self._latest_by_asset(session)
-            # One query, not one per asset: the reviewer needs to see a divergent
-            # checkpoint, which is neither "verified" nor a plain failure (US07-03).
-            projections = {
-                row.asset_id: row.state
-                for row in session.scalars(
-                    select(ExifProjection).where(ExifProjection.stage == "safety")
-                )
-            }
             rows = []
             for asset in assets:
                 review = latest.get(asset.id)
@@ -196,7 +189,6 @@ class SafetyService:
                         "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),
-                        "exif_state": projections.get(asset.id),
                     }
                 )
             return {"total": len(rows), "items": rows[offset : offset + limit]}
@@ -255,24 +247,13 @@ class SafetyService:
         result_sha256 = None
         if write_exif and decision in (SFW, NSFW) and path:
             ops = exif_projection(decision)
-            # The full checkpoint: write the owned keyword, read the whole file back,
-            # and prove every field this stage does not own survived. A divergent
-            # result is recorded and left alone — it must not count as verified, so
-            # upload stays blocked until a human decides (US07-03).
-            result = exif_checkpoint.run(
-                path, add=tuple(ops["add"]), remove=tuple(ops["remove"])
-            )
-            exif_checkpoint.record(
-                self._session_factory,
-                asset_id=asset_id,
-                stage="safety",
-                result=result,
-                add=tuple(ops["add"]),
-                remove=tuple(ops["remove"]),
-            )
-            if result.verified:
-                exif_verified_at = result.verified_at
-                result_sha256 = result.sha256
+            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:

photo_pipeline/services/thumbnails.py

@@ -20,16 +20,15 @@ pa-imaging).
 
 from __future__ import annotations
 
-import os
 import uuid
 from pathlib import Path
 
-from PIL import Image, ImageOps
+from PIL import Image, ImageOps, UnidentifiedImageError
 from sqlalchemy import select
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import sessionmaker
 
-from photo_pipeline import imaging, path_policy
+from photo_pipeline import path_policy
 from photo_pipeline.config import Config
 from photo_pipeline.models import Asset, Thumbnail
 from photo_pipeline.services import availability
@@ -43,9 +42,6 @@ except Exception:  # pragma: no cover
     pass
 
 SIZES = (256, 512, 1280)
-# Every in-flight render writes one of these beside its destination; startup
-# cleanup recognises exactly this pattern and nothing else.
-TEMP_SUFFIX = ".tmp"
 THUMB_VERSION = 1
 THUMB_FORMAT = "webp"
 # The size kept as durable comparison evidence for archived assets (concept §9).
@@ -226,33 +222,28 @@ class ThumbnailService:
         return self._cache_dir / safe[:2] / f"{safe}.{THUMB_FORMAT}"
 
     def _render(self, source: str, size: int, cache_key: str) -> dict:
-        """Decode bounded, orient, colour-manage, resize, and write atomically.
-
-        The temporary file is removed on every failure path: a decoder that dies
-        halfway through ``save`` would otherwise leave a stray ``.tmp`` in the cache
-        forever (US07-03).
-        """
-        destination = self._cache_path(cache_key)
-        destination.parent.mkdir(parents=True, exist_ok=True)
-        tmp = destination.with_name(f".{destination.name}.{uuid.uuid4().hex}{TEMP_SUFFIX}")
         try:
-            with imaging.open_image(source, max_pixels=self._config.thumbnail_max_pixels) as image:
-                imaging.draft(image, size)  # JPEG decodes near the target size
+            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 = imaging.to_srgb(oriented, mode=mode)
+                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 imaging.ImageTooLarge as error:
-            tmp.unlink(missing_ok=True)
-            raise ImageTooLarge(str(error)) from None
-        except imaging.UndecodableImage as error:
-            tmp.unlink(missing_ok=True)
-            raise UnsupportedImage(str(error)) from None
-        except Exception:
-            tmp.unlink(missing_ok=True)
+        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 {
@@ -262,28 +253,6 @@ class ThumbnailService:
             "format": THUMB_FORMAT,
         }
 
-    def cleanup_temp_files(self) -> int:
-        """Remove leftover render temporaries, and only those.
-
-        Runs at startup, where the concept is explicit: clean *recognized* stale
-        temporary files, never arbitrary ones. The pattern is this service's own —
-        a dot-prefixed name inside the managed cache directory ending in
-        ``.tmp`` — so nothing outside the cache and nothing a user put there can
-        match. Returns how many were removed.
-        """
-        if not self._cache_dir.is_dir():
-            return 0
-        removed = 0
-        for leftover in self._cache_dir.rglob(f".*{TEMP_SUFFIX}"):
-            if not leftover.is_file() or leftover.is_symlink():
-                continue
-            try:
-                leftover.unlink()
-                removed += 1
-            except OSError:  # pragma: no cover - a racing render already won
-                pass
-        return removed
-
     # ── persistence ────────────────────────────────────────────────────────────
     def _record_ready(
         self, cache_key: str, asset_id: str, size: int, rendered: dict, *, protected: bool = False

photo_pipeline/services/upload_batches.py

@@ -37,7 +37,6 @@ from sqlalchemy import select, update
 from sqlalchemy.orm import sessionmaker
 
 from photo_pipeline.config import Config
-from photo_pipeline.faults import UPLOAD_ACCEPTED, maybe_fault
 from photo_pipeline.integrations import immich_go
 from photo_pipeline.models import UploadBatch, UploadItem
 from photo_pipeline.services.hashing import sha1_file
@@ -237,10 +236,6 @@ class UploadBatchService:
             error = ("uploader_failed", f"immich-go exited with {result['exit_code']}")
             item_state = ItemState.FAILED
 
-        # The uploader is done and Immich may already hold every file, but nothing
-        # about that is durable yet — the control point for "accepted, outcome not
-        # recorded" (US07-04). Recovery must answer ``unknown_requires_verification``.
-        maybe_fault(UPLOAD_ACCEPTED)
         self._finish(batch_id, token=token, state=state, error=error, result=result)
         if item_state:
             self._set_items(batch_id, item_state)

tests/_artifacts.py

@@ -1,96 +0,0 @@
-"""Failure artifacts for the fault and race suites (US07-04).
-
-A randomized concurrency failure that leaves nothing behind is a failure nobody
-can diagnose: the temporary library is deleted, the database goes with it, and the
-seed that produced the interleaving is gone. So when a test fails, everything
-needed to reproduce and read it is copied out of the temporary directory:
-
-    <artifacts>/<test id>/
-        seeds.json     recorded properties (``race_seed``) and the failing test id
-        manifest.json  every file under the temporary directory: path, size, sha256
-        files/...      the databases (with -wal/-shm), journals, and logs themselves
-
-The manifest covers the whole tree — including files too large or too private to
-copy — so a missing or unexpected file is still visible afterwards. Copying is
-bounded by ``MAX_COPY_BYTES``: artifacts must not turn a failing CI run into a
-disk-full one.
-
-Set ``PHOTO_PIPELINE_TEST_ARTIFACTS`` to choose the destination; the default is
-``.artifacts/`` in the repository root.
-"""
-
-from __future__ import annotations
-
-import hashlib
-import json
-import os
-import re
-import shutil
-from pathlib import Path
-
-REPO = Path(__file__).resolve().parents[1]
-DEFAULT_DIR = REPO / ".artifacts"
-MAX_COPY_BYTES = 25 * 1024 * 1024
-# Databases (and their write-ahead logs), journals exported as files, and logs.
-COPY_SUFFIXES = (".db", ".db-wal", ".db-shm", ".sqlite", ".log", ".json", ".jsonl", ".argv")
-
-
-def artifacts_dir() -> Path:
-    return Path(os.environ.get("PHOTO_PIPELINE_TEST_ARTIFACTS", DEFAULT_DIR))
-
-
-def _slug(test_id: str) -> str:
-    return re.sub(r"[^A-Za-z0-9._-]+", "_", test_id)[:120]
-
-
-def manifest(root: Path) -> list[dict]:
-    """Every file under ``root``: relative path, byte size, and SHA-256.
-
-    The filesystem state at the moment of failure — what was moved, what was left
-    behind, what was half-written.
-    """
-    entries = []
-    for path in sorted(root.rglob("*")):
-        if not path.is_file() or path.is_symlink():
-            continue
-        try:
-            body = path.read_bytes()
-        except OSError as error:
-            entries.append({"path": str(path.relative_to(root)), "error": str(error)})
-            continue
-        entries.append(
-            {
-                "path": str(path.relative_to(root)),
-                "bytes": len(body),
-                "sha256": hashlib.sha256(body).hexdigest(),
-            }
-        )
-    return entries
-
-
-def collect(root: Path, test_id: str, *, properties: dict | None = None) -> Path:
-    """Copy the evidence for one failed test out of ``root``. Returns its directory."""
-    destination = artifacts_dir() / _slug(test_id)
-    files = destination / "files"
-    files.mkdir(parents=True, exist_ok=True)
-
-    entries = manifest(root)
-    (destination / "manifest.json").write_text(json.dumps(entries, indent=1))
-    (destination / "seeds.json").write_text(
-        json.dumps({"test": test_id, "properties": properties or {}}, indent=1)
-    )
-
-    budget = MAX_COPY_BYTES
-    for path in sorted(root.rglob("*")):
-        if not path.is_file() or path.is_symlink():
-            continue
-        if not path.name.endswith(COPY_SUFFIXES):
-            continue
-        size = path.stat().st_size
-        if size > budget:
-            continue  # the manifest still records it; the copy is what is skipped
-        target = files / path.relative_to(root)
-        target.parent.mkdir(parents=True, exist_ok=True)
-        shutil.copy2(path, target)
-        budget -= size
-    return destination

tests/conftest.py

@@ -1,6 +1,5 @@
-"""Make the repository root importable for the pipeline test suites, give every
-suite the application session the API requires since US07-02, and keep the evidence
-of any failure (US07-04).
+"""Make the repository root importable for the pipeline test suites, and give every
+suite the application session the API requires since US07-02.
 
 The suites drive the API the way the browser does — module-level ``httpx`` calls and
 ``TestClient`` — so instead of threading a cookie through several hundred call sites,
@@ -100,30 +99,3 @@ def _api_session():
     httpx._api.request, httpx.request = real_request, real_request
     httpx._api.stream, httpx.stream = real_stream, real_stream
     TestClient.request, TestClient.__init__ = real_client_request, real_client_init
-
-
-@pytest.hookimpl(hookwrapper=True)
-def pytest_runtest_makereport(item, call):
-    """On failure, keep the temporary library, database, journals, logs, and seed.
-
-    ``tmp_path`` is deleted a few runs later and its contents are the only record
-    of what a randomized or fault-injected failure actually did, so they are copied
-    out while they still exist (US07-04).
-    """
-    outcome = yield
-    report = outcome.get_result()
-    if report.when != "call" or not report.failed:
-        return
-    root = item.funcargs.get("tmp_path")
-    if root is None or not Path(root).exists():
-        return
-    from tests._artifacts import collect
-
-    try:
-        destination = collect(
-            Path(root), item.nodeid, properties=dict(getattr(report, "user_properties", []))
-        )
-    except OSError as error:  # never let evidence collection mask the real failure
-        report.sections.append(("failure artifacts", f"could not be collected: {error}"))
-    else:
-        report.sections.append(("failure artifacts", str(destination)))

tests/e2e/test_crash_recovery.py

@@ -1,274 +0,0 @@
-"""Process death at the newer control points (US07-04, concept §18).
-
-The rename and archive journals already prove crash safety at each of their
-transitions (tests/integration/test_rename_recovery.py,
-tests/integration/test_archive_recovery.py). The three transitions covered here
-are the remaining ones where a kill leaves the world and the database disagreeing:
-
-- ``exif:written``   — keywords are on disk, nothing about them is recorded;
-- ``upload:accepted``— the uploader finished, no outcome is stored;
-- ``job:item_done``  — one item is durably done, the job is not finished.
-
-Each test kills a real child process at the barrier and then asserts what a
-restart does: resume idempotently, or say plainly that a human has to look. Never
-"assume it worked".
-"""
-
-from __future__ import annotations
-
-import os
-import shutil
-import subprocess
-import sys
-import uuid
-from datetime import datetime, timezone
-from pathlib import Path
-
-import numpy as np
-import pytest
-from PIL import Image
-from sqlalchemy import select
-
-from photo_pipeline.config import Config
-from photo_pipeline.db import create_db_engine, create_session_factory, run_migrations
-from photo_pipeline.faults import EXIF_WRITTEN, JOB_ITEM_DONE, UPLOAD_ACCEPTED
-from photo_pipeline.integrations import exiftool
-from photo_pipeline.jobs.worker import Worker
-from photo_pipeline.models import Asset, SafetyReview
-from photo_pipeline.services import exif_checkpoint, hashing
-from photo_pipeline.services.jobs import ItemState, JobService, JobState
-from photo_pipeline.services.safety import SafetyService
-from photo_pipeline.services.upload_batches import BatchState, UploadBatchService
-from photo_pipeline.services.upload_verification import retry_blockers
-from photo_pipeline.services.uploads import UploadService
-from tests.e2e._pipeline_harness import (
-    SILENT_UPLOADER,
-    FakeImmich,
-    fake_uploader,
-    mark_upload_ready,
-    seed_album,
-)
-
-REPO = Path(__file__).resolve().parents[2]
-NOW = datetime(2026, 1, 1, tzinfo=timezone.utc)
-
-
-def _child(script: str, *args: str, barrier: str, tmp_path: Path) -> None:
-    """Run ``script`` in a child that dies at ``barrier``; assert it really died."""
-    path = tmp_path / f"child_{barrier.replace(':', '_')}.py"
-    path.write_text(script.format(repo=str(REPO)))
-    env = dict(os.environ)
-    env["PHOTO_PIPELINE_FAULT_AFTER"] = barrier
-    result = subprocess.run(
-        [sys.executable, str(path), *args], env=env, capture_output=True
-    )
-    assert result.returncode in (9, -9), (
-        f"child should have been killed at {barrier}, got {result.returncode}: "
-        f"{result.stderr.decode(errors='replace')[-400:]}"
-    )
-
-
-def _env(tmp_path, **extra):
-    (tmp_path / "data").mkdir(exist_ok=True)
-    lib = tmp_path / "lib"
-    lib.mkdir(exist_ok=True)
-    config = Config.from_env(
-        {
-            "PHOTO_PIPELINE_DATA_DIR": str(tmp_path / "data"),
-            "PHOTO_PIPELINE_LIBRARY_ROOTS": str(lib),
-            **extra,
-        }
-    )
-    run_migrations(config.database_url)
-    return config, create_session_factory(create_db_engine(config.database_url)), lib
-
-
-def _image(path: Path, seed: int = 3) -> None:
-    path.parent.mkdir(parents=True, exist_ok=True)
-    pixels = np.random.default_rng(seed).integers(0, 256, (64, 96, 3), dtype=np.uint8)
-    Image.fromarray(pixels).save(path, quality=90)
-
-
-def _register(sf, path: Path) -> str:
-    asset_id = str(uuid.uuid4())
-    with sf() as session:
-        session.add(
-            Asset(
-                id=asset_id,
-                original_path=str(path),
-                current_path=str(path),
-                discovered_at=NOW,
-                hash_version=1,
-                byte_size=path.stat().st_size,
-                current_sha256=hashing.sha256_file(path),
-            )
-        )
-        session.commit()
-    return asset_id
-
-
-# ── EXIF written, checkpoint not recorded ────────────────────────────────────
-
-DECIDE_SCRIPT = """
-import sys
-sys.path.insert(0, {repo!r})
-from photo_pipeline.db import create_db_engine, create_session_factory
-from photo_pipeline.services.safety import SafetyService
-
-db_url, asset_id = sys.argv[1], sys.argv[2]
-sf = create_session_factory(create_db_engine(db_url))
-SafetyService(sf).decide(asset_id, "nsfw")
-"""
-
-
-@pytest.mark.skipif(shutil.which("exiftool") is None, reason="exiftool not installed")
-def test_a_crash_after_the_exif_write_leaves_nothing_verified_and_re_runs_cleanly(tmp_path):
-    config, sf, lib = _env(tmp_path)
-    path = lib / "a.jpg"
-    _image(path)
-    asset_id = _register(sf, path)
-    original_sha = hashing.sha256_file(path)
-
-    _child(
-        DECIDE_SCRIPT, config.database_url, asset_id, barrier=EXIF_WRITTEN, tmp_path=tmp_path
-    )
-
-    # The file changed, but the application claims nothing about it: no decision,
-    # no projection, and the stored hash is still the pre-write one.
-    assert "nsfw" in exiftool.read_keyword_sets([str(path)])[str(path)]
-    assert hashing.sha256_file(path) != original_sha
-    with sf() as session:
-        assert session.scalars(select(SafetyReview)).all() == []
-        assert session.get(Asset, asset_id).current_sha256 == original_sha
-    assert exif_checkpoint.state_for(sf, asset_id, "safety") is None
-
-    # Re-running is the recovery: the write is idempotent, so the second attempt
-    # verifies and records what the first one only did to the file.
-    review = SafetyService(sf).decide(asset_id, "nsfw")
-    assert review["exif_verified"] is True
-    assert exif_checkpoint.state_for(sf, asset_id, "safety") == exif_checkpoint.VERIFIED
-    keywords = exiftool.read_keyword_sets([str(path)])[str(path)]
-    assert "nsfw" in keywords and "sfw" not in keywords
-    with sf() as session:
-        asset = session.get(Asset, asset_id)
-        assert asset.current_sha256 == hashing.sha256_file(path)
-
-
-# ── uploader accepted, outcome not persisted ─────────────────────────────────
-
-UPLOAD_SCRIPT = """
-import sys
-sys.path.insert(0, {repo!r})
-from photo_pipeline.config import Config
-from photo_pipeline.db import create_db_engine, create_session_factory
-from photo_pipeline.services.upload_batches import UploadBatchService
-
-db_url, data_dir, lib, binary, server, batch_id = sys.argv[1:7]
-config = Config.from_env(
-    {{
-        "PHOTO_PIPELINE_DATA_DIR": data_dir,
-        "PHOTO_PIPELINE_LIBRARY_ROOTS": lib,
-        "PHOTO_PIPELINE_IMMICH_GO_BINARY": binary,
-        "PHOTO_PIPELINE_IMMICH_SERVER_URL": server,
-        "PHOTO_PIPELINE_IMMICH_API_KEY": "sentinel",
-    }}
-)
-sf = create_session_factory(create_db_engine(db_url))
-UploadBatchService(sf, config=config).run(batch_id)
-"""
-
-
-def test_a_crash_after_the_uploader_accepted_requires_verification(tmp_path):
-    """immich-go exited cleanly and the server may hold every file, but nothing was
-    written down. Recovery must not guess success — and must not blindly retry."""
-    seeded = seed_album(tmp_path)
-    mark_upload_ready(seeded)
-    immich = FakeImmich()
-    binary = fake_uploader(tmp_path, SILENT_UPLOADER)
-    config = Config.from_env(
-        {
-            "PHOTO_PIPELINE_DATA_DIR": str(seeded.data),
-            "PHOTO_PIPELINE_LIBRARY_ROOTS": str(seeded.lib),
-            "PHOTO_PIPELINE_IMMICH_GO_BINARY": str(binary),
-            "PHOTO_PIPELINE_IMMICH_SERVER_URL": immich.url,
-            "PHOTO_PIPELINE_IMMICH_API_KEY": "sentinel",
-        }
-    )
-    engine = create_db_engine(config.database_url)
-    sf = create_session_factory(engine)
-    service = UploadBatchService(sf, config=config)
-    report = UploadService(sf, config=config).preflight(["rome"])
-    assert report["state"] == "ready", report["blockers"]
-    batch_id = service.create(["rome"], token=report["token"])[0]["id"]
-
-    try:
-        # The batch is claimed by the child, which dies once the uploader has run.
-        _child(
-            UPLOAD_SCRIPT,
-            config.database_url,
-            str(seeded.data),
-            str(seeded.lib),
-            str(binary),
-            immich.url,
-            batch_id,
-            barrier=UPLOAD_ACCEPTED,
-            tmp_path=tmp_path,
-        )
-    finally:
-        immich.stop()
-    assert service.get(batch_id)["state"] == BatchState.RUNNING  # lane still held
-
-    recovered = service.recover()
-
-    assert recovered == {"interrupted": 1}
-    batch = service.get(batch_id)
-    assert batch["state"] == BatchState.UNKNOWN and batch["error_code"] == "interrupted"
-    assert [b["code"] for b in retry_blockers(batch)] == ["requires_verification"]
-    assert all(item["state"] == "pending" for item in batch["items"]), "nothing claimed as sent"
-
-
-# ── one item done, the job unfinished ────────────────────────────────────────
-
-WORKER_SCRIPT = """
-import sys
-sys.path.insert(0, {repo!r})
-from pathlib import Path
-from photo_pipeline.db import create_db_engine, create_session_factory
-from photo_pipeline.jobs.worker import Worker
-
-db_url, log = sys.argv[1], Path(sys.argv[2])
-
-def handler(item_key, ctx):
-    with log.open("a") as handle:
-        handle.write(item_key + "\\n")
-
-sf = create_session_factory(create_db_engine(db_url))
-Worker(sf, {{"scan": handler}}, "killable", lease_seconds=1).run_once()
-"""
-
-
-def test_a_crash_between_items_resumes_without_running_the_done_item_twice(tmp_path):
-    config, sf, lib = _env(tmp_path)
-    service = JobService(sf)
-    job = service.enqueue("scan", items=["a", "b", "c"])
-    log = tmp_path / "handled.log"
-
-    _child(
-        WORKER_SCRIPT, config.database_url, str(log), barrier=JOB_ITEM_DONE, tmp_path=tmp_path
-    )
-
-    crashed = log.read_text().split()
-    assert crashed == ["a"], "the child should have died right after its first item"
-    assert service.get(job["id"])["state"] == JobState.RUNNING
-
-    # A fresh worker takes over once the dead lease expires.
-    import time
-
-    time.sleep(1.1)  # the child's lease is one second long
-    fresh = Worker(sf, {"scan": lambda item, ctx: log.open("a").write(item + "\n")}, "alive")
-    fresh.run_once()
-
-    assert service.get(job["id"])["state"] == JobState.SUCCEEDED
-    handled = log.read_text().split()
-    assert sorted(handled) == ["a", "b", "c"], f"an item ran twice or not at all: {handled}"
-    assert service.progress(job["id"])["by_state"] == {ItemState.SUCCEEDED: 3}

tests/fixtures/media_corpus.py

@@ -1,325 +0,0 @@
-"""The golden media corpus: every format, orientation, profile, damage, and
-metadata case this application claims to survive (US07-03, concept §18).
-
-``CASES`` is the manifest and the authority. Each entry declares a stable logical
-id (never a path), how the file is generated, and what the pipeline must do with
-it — decode it, or refuse it with one precise error code. Tests parametrize over the
-manifest, so a case that is added here without an expectation, or an expectation
-that stops holding, fails the suite rather than quietly going untested.
-
-Everything is generated, never committed: fixed pixel seeds, fixed EXIF strings, no
-clock, no network, no personal data. Regeneration is byte-stable, which
-``test_media_hardening.py`` proves by building the corpus twice and comparing
-checksums — a golden corpus that drifts is not golden.
-"""
-
-from __future__ import annotations
-
-import struct
-import zlib
-from dataclasses import dataclass, field
-from pathlib import Path
-from typing import Callable
-
-import numpy as np
-from PIL import Image
-
-# Error codes the pipeline may answer with; ``None`` means "must render".
-UNSUPPORTED = "unsupported_image"
-TOO_LARGE = "image_too_large"
-
-# One fixed capture time for every metadata case: the corpus must not depend on when
-# it was generated.
-CAPTURE_TIME = "2019:07:14 10:30:00"
-
-
-@dataclass(frozen=True)
-class MediaCase:
-    id: str
-    filename: str
-    kind: str  # format | orientation | profile | damaged | metadata
-    build: Callable[[Path], None]
-    expect_error: str | None = None
-    # Declared for the metadata cases: exiftool arguments applied after the pixels
-    # are written, and the user fields that must survive every later stage.
-    exif_args: tuple[str, ...] = ()
-    preserved_fields: tuple[str, ...] = ()
-    notes: str = ""
-    tags: tuple[str, ...] = field(default_factory=tuple)
-
-
-# ── generators ────────────────────────────────────────────────────────────────
-
-
-def _pixels(width: int, height: int, seed: int, bands: int = 3) -> np.ndarray:
-    return np.random.default_rng(seed).integers(0, 256, (height, width, bands), dtype=np.uint8)
-
-
-def _save(path: Path, image: Image.Image, **kwargs) -> None:
-    path.parent.mkdir(parents=True, exist_ok=True)
-    image.save(path, **kwargs)
-
-
-def _jpeg(width=320, height=240, seed=1, **kwargs):
-    def build(path: Path) -> None:
-        _save(path, Image.fromarray(_pixels(width, height, seed)), quality=90, **kwargs)
-
-    return build
-
-
-def _oriented_jpeg(orientation: int):
-    def build(path: Path) -> None:
-        image = Image.fromarray(_pixels(400, 200, 5))  # landscape source
-        exif = image.getexif()
-        exif[274] = orientation  # 0x0112 Orientation
-        _save(path, image, exif=exif, quality=90)
-
-    return build
-
-
-def _rotated_pixels(path: Path) -> None:
-    """The same scene rotated in the pixels instead of in a tag."""
-    _save(path, Image.fromarray(_pixels(400, 200, 5)).transpose(Image.ROTATE_90), quality=90)
-
-
-def _png(alpha: bool = False, seed: int = 2):
-    def build(path: Path) -> None:
-        if alpha:
-            _save(path, Image.fromarray(_pixels(120, 90, seed, bands=4), "RGBA"))
-        else:
-            _save(path, Image.fromarray(_pixels(120, 90, seed)))
-
-    return build
-
-
-def _webp(path: Path) -> None:
-    _save(path, Image.fromarray(_pixels(150, 100, 3)), quality=80)
-
-
-def _tiff(path: Path) -> None:
-    _save(path, Image.fromarray(_pixels(140, 110, 4)))
-
-
-def _grayscale(path: Path) -> None:
-    _save(path, Image.fromarray(_pixels(100, 100, 6)).convert("L"), quality=90)
-
-
-def _cmyk(path: Path) -> None:
-    _save(path, Image.fromarray(_pixels(100, 100, 7)).convert("CMYK"), quality=90)
-
-
-def _tiny(path: Path) -> None:
-    _save(path, Image.fromarray(_pixels(1, 1, 8)))
-
-
-def _icc_tagged(path: Path) -> None:
-    """A profile-bearing image: the colour-managed decode path must run."""
-    from PIL import ImageCms
-
-    profile = bytearray(ImageCms.ImageCmsProfile(ImageCms.createProfile("sRGB")).tobytes())
-    # An ICC header carries its creation timestamp at bytes 24..35. Left alone, the
-    # corpus would be a different corpus every time it is generated.
-    profile[24:36] = b"\x00" * 12
-    _save(path, Image.fromarray(_pixels(120, 80, 9)), icc_profile=bytes(profile), quality=90)
-
-
-def _broken_icc(path: Path) -> None:
-    """A profile that is not a profile: a picture is still a picture."""
-    _save(path, Image.fromarray(_pixels(120, 80, 10)), icc_profile=b"not-a-profile", quality=90)
-
-
-def _wide_jpeg(path: Path) -> None:
-    """Large enough that decoding it at full resolution is visible in memory."""
-    _save(path, Image.fromarray(_pixels(4000, 3000, 11)), quality=70)
-
-
-def _zero_byte(path: Path) -> None:
-    path.parent.mkdir(parents=True, exist_ok=True)
-    path.write_bytes(b"")
-
-
-def _truncated_jpeg(path: Path) -> None:
-    image = Image.fromarray(_pixels(400, 300, 12))
-    _save(path, image, quality=90)
-    data = path.read_bytes()
-    path.write_bytes(data[: len(data) // 2])  # header intact, pixels missing
-
-
-def _corrupt_png(path: Path) -> None:
-    image = Image.fromarray(_pixels(120, 90, 13))
-    _save(path, image)
-    data = bytearray(path.read_bytes())
-    data[40:80] = b"\x00" * 40  # shred the compressed stream, keep the header
-    path.write_bytes(bytes(data))
-
-
-def _not_an_image(path: Path) -> None:
-    path.parent.mkdir(parents=True, exist_ok=True)
-    path.write_bytes(b"This is a text file that happens to be named .jpg\n")
-
-
-def _png_declaring(width: int, height: int):
-    """A tiny, structurally valid PNG whose header claims an enormous picture.
-
-    A few hundred bytes on disk, gigapixels on paper: the pipeline must refuse it
-    from the declared dimensions, before a single pixel is allocated. Rewriting the
-    IHDR of a real PNG (rather than hand-rolling a stub) keeps the file openable, so
-    the refusal is proven to come from the size check and not from a parse failure.
-    """
-
-    def build(path: Path) -> None:
-        import io
-
-        buffer = io.BytesIO()
-        Image.fromarray(_pixels(4, 4, 15)).save(buffer, format="PNG")
-        data = bytearray(buffer.getvalue())
-        start = 8 + 4  # PNG signature, then the IHDR length field
-        struct.pack_into(">II", data, start + 4, width, height)
-        ihdr = bytes(data[start : start + 4 + 13])
-        struct.pack_into(">I", data, start + 4 + 13, zlib.crc32(ihdr))
-        path.parent.mkdir(parents=True, exist_ok=True)
-        path.write_bytes(bytes(data))
-
-    return build
-
-
-CASES: tuple[MediaCase, ...] = (
-    # ── formats ───────────────────────────────────────────────────────────────
-    MediaCase("jpeg", "formats/plain.jpg", "format", _jpeg()),
-    MediaCase("jpeg_uppercase_ext", "formats/UPPER.JPG", "format", _jpeg(seed=14)),
-    MediaCase("png", "formats/plain.png", "format", _png()),
-    MediaCase("png_alpha", "formats/alpha.png", "format", _png(alpha=True)),
-    MediaCase("webp", "formats/plain.webp", "format", _webp),
-    MediaCase("tiff", "formats/plain.tiff", "format", _tiff),
-    # ── orientation ───────────────────────────────────────────────────────────
-    *(
-        MediaCase(
-            f"orientation_{value}",
-            f"orientation/exif_{value}.jpg",
-            "orientation",
-            _oriented_jpeg(value),
-            notes="EXIF orientation must be applied before resizing",
-        )
-        for value in range(1, 9)
-    ),
-    MediaCase("rotated_pixels", "orientation/rotated.jpg", "orientation", _rotated_pixels),
-    # ── colour and size profiles ──────────────────────────────────────────────
-    MediaCase("grayscale", "profiles/gray.jpg", "profile", _grayscale),
-    MediaCase("cmyk", "profiles/cmyk.jpg", "profile", _cmyk),
-    MediaCase("tiny", "profiles/tiny.png", "profile", _tiny),
-    MediaCase("icc_tagged", "profiles/icc.jpg", "profile", _icc_tagged),
-    MediaCase(
-        "broken_icc",
-        "profiles/broken-icc.jpg",
-        "profile",
-        _broken_icc,
-        notes="an unusable ICC profile falls back to a plain conversion, never an error",
-    ),
-    MediaCase(
-        "large_jpeg",
-        "profiles/large.jpg",
-        "profile",
-        _wide_jpeg,
-        notes="12 megapixels: the decode must stay near the requested size",
-    ),
-    # ── damaged and hostile inputs ────────────────────────────────────────────
-    MediaCase("zero_byte", "damaged/empty.jpg", "damaged", _zero_byte, UNSUPPORTED),
-    MediaCase("truncated_jpeg", "damaged/truncated.jpg", "damaged", _truncated_jpeg, UNSUPPORTED),
-    MediaCase("corrupt_png", "damaged/corrupt.png", "damaged", _corrupt_png, UNSUPPORTED),
-    MediaCase("text_as_jpeg", "damaged/text.jpg", "damaged", _not_an_image, UNSUPPORTED),
-    MediaCase(
-        "bomb_header",
-        "damaged/bomb.png",
-        "damaged",
-        _png_declaring(60_000, 60_000),
-        TOO_LARGE,
-        notes="3.6 gigapixels declared in the header and nothing else",
-    ),
-    MediaCase(
-        "bomb_warning_band",
-        "damaged/bomb-warning.png",
-        "damaged",
-        _png_declaring(10_000, 10_000),
-        TOO_LARGE,
-        notes="inside Pillow's warn-only band; the warning is promoted to a refusal",
-    ),
-    # ── metadata ──────────────────────────────────────────────────────────────
-    MediaCase("no_exif", "metadata/bare.jpg", "metadata", _jpeg(seed=20)),
-    MediaCase(
-        "user_exif",
-        "metadata/user.jpg",
-        "metadata",
-        _jpeg(seed=21),
-        exif_args=(
-            "-Artist=Ada Lovelace",
-            "-Copyright=(c) Ada",
-            f"-DateTimeOriginal={CAPTURE_TIME}",
-            "-GPSLatitude=48.137",
-            "-GPSLatitudeRef=N",
-            "-Rating=4",
-            "-ImageDescription=A day out",
-        ),
-        preserved_fields=(
-            "EXIF:IFD0:Artist",
-            "EXIF:IFD0:Copyright",
-            "EXIF:ExifIFD:DateTimeOriginal",
-            "EXIF:IFD0:ImageDescription",
-            "XMP:XMP-xmp:Rating",
-        ),
-        notes="user metadata that every stage must leave exactly as it found it",
-    ),
-    MediaCase(
-        "prior_safety_keyword",
-        "metadata/prior-safety.jpg",
-        "metadata",
-        _jpeg(seed=22),
-        exif_args=("-Keywords+=nsfw", "-Subject+=nsfw", "-Artist=Ada Lovelace"),
-        preserved_fields=("EXIF:IFD0:Artist",),
-        notes="a safety decision already written by an earlier run",
-    ),
-    MediaCase(
-        "prior_analysis_keywords",
-        "metadata/prior-analysis.jpg",
-        "metadata",
-        _jpeg(seed=23),
-        exif_args=("-Keywords+=beach", "-Keywords+=sunset", "-Subject+=beach", "-Subject+=sunset"),
-        notes="analysis keywords from an earlier run; a safety write must not drop them",
-    ),
-    MediaCase(
-        "conflicting_safety_keywords",
-        "metadata/conflicting.jpg",
-        "metadata",
-        _jpeg(seed=24),
-        exif_args=("-Keywords+=sfw", "-Keywords+=nsfw", "-Subject+=sfw", "-Subject+=nsfw"),
-        notes="both safety keywords at once: mutually exclusive means one must go",
-    ),
-    MediaCase(
-        "malformed_metadata",
-        "metadata/malformed.jpg",
-        "metadata",
-        _jpeg(seed=25, exif=b"\x00\x01\x02not-a-valid-exif-block"),
-        notes="a broken EXIF block must not stop the picture from being usable",
-    ),
-)
-
-CASES_BY_ID = {case.id: case for case in CASES}
-
-
-def build_corpus(root: Path, *, ids: tuple[str, ...] | None = None) -> dict[str, Path]:
-    """Generate the corpus (or a named subset) under ``root``; return id → path."""
-    import subprocess
-
-    built: dict[str, Path] = {}
-    for case in CASES:
-        if ids is not None and case.id not in ids:
-            continue
-        path = root / case.filename
-        case.build(path)
-        if case.exif_args:
-            subprocess.run(
-                ["exiftool", "-m", "-overwrite_original", *case.exif_args, str(path)],
-                capture_output=True,
-                check=False,
-            )
-        built[case.id] = path
-    return built

tests/integration/test_concurrency_races.py

@@ -1,535 +0,0 @@
-"""Randomized concurrency and race tests (US07-04, concept §16 and §18).
-
-Every test here runs several times with a *recorded* seed: the seed decides the
-jitter injected around each racing operation, it is attached to the test result
-(``race_seed``), and a failing run can be replayed exactly with
-
-    PHOTO_PIPELINE_RACE_SEED=<seed> pytest tests/integration/test_concurrency_races.py
-
-``PHOTO_PIPELINE_RACE_REPEATS`` raises the repeat count for a soak run; the
-default is small enough to belong in the ordinary suite.
-
-The assertions are invariants, not schedules — a race whose interleaving decides
-the *outcome* is fine, one whose interleaving decides whether the database still
-makes sense is not:
-
-- no work is claimed, executed, or completed twice;
-- no commit from a stale fencing token lands;
-- no file is lost, overwritten, or left with foreign content;
-- a decision that changed mid-flight is never overwritten by the older answer;
-- the database passes ``PRAGMA integrity_check`` afterwards.
-"""
-
-from __future__ import annotations
-
-import os
-import random
-import subprocess
-import sys
-import threading
-import time
-import uuid
-from collections import Counter
-from concurrent.futures import ThreadPoolExecutor
-from datetime import datetime, timedelta, timezone
-from pathlib import Path
-
-import numpy as np
-import pytest
-from PIL import Image
-from sqlalchemy import select, text
-
-from photo_pipeline.config import Config
-from photo_pipeline.db import create_db_engine, create_session_factory, run_migrations
-from photo_pipeline.faults import JOB_ITEM_DONE
-from photo_pipeline.jobs.worker import Worker
-from photo_pipeline.models import (
-    AlbumProposal,
-    AnalysisResult,
-    Asset,
-    ExifProjection,
-    SafetyReview,
-)
-from photo_pipeline.services import hashing
-from photo_pipeline.services.analysis import AnalysisService
-from photo_pipeline.services.jobs import (
-    InvalidTransition,
-    ItemState,
-    JobConflict,
-    JobService,
-    JobState,
-    _now,
-)
-from photo_pipeline.services.rename_apply import RenameApplyService
-from photo_pipeline.services.renames import RenameService
-from photo_pipeline.services.thumbnails import ThumbnailError, ThumbnailService
-
-REPO = Path(__file__).resolve().parents[2]
-NOW = datetime(2026, 1, 1, tzinfo=timezone.utc)
-REPEATS = int(os.environ.get("PHOTO_PIPELINE_RACE_REPEATS", "3"))
-MAX_JITTER_SECONDS = 0.01
-
-
-# ── seeded randomness ────────────────────────────────────────────────────────
-
-
-@pytest.fixture(params=range(REPEATS), ids=lambda run: f"run{run}")
-def rng(request, record_property):
-    """A seeded RNG whose seed is recorded on the test result.
-
-    Without a pinned seed the run is genuinely random — which is the point, a
-    fixed schedule stops finding new interleavings after the first green run —
-    so the seed is reported for replay instead.
-    """
-    pinned = os.environ.get("PHOTO_PIPELINE_RACE_SEED")
-    seed = int(pinned) + request.param if pinned else random.SystemRandom().randrange(2**32)
-    record_property("race_seed", seed)
-    print(f"race seed: {seed}")  # visible with -s and in the failure report
-    return random.Random(seed)
-
-
-def jitter(rng: random.Random) -> None:
-    """Sleep a random sliver so racing threads interleave differently each run."""
-    time.sleep(rng.uniform(0, MAX_JITTER_SECONDS))
-
-
-# ── fixtures ─────────────────────────────────────────────────────────────────
-
-
-def _config(tmp_path) -> tuple[Config, Path]:
-    (tmp_path / "data").mkdir(exist_ok=True)
-    lib = tmp_path / "lib"
-    lib.mkdir(exist_ok=True)
-    config = Config.from_env(
-        {
-            "PHOTO_PIPELINE_DATA_DIR": str(tmp_path / "data"),
-            "PHOTO_PIPELINE_LIBRARY_ROOTS": str(lib),
-        }
-    )
-    run_migrations(config.database_url)
-    return config, lib
-
-
-@pytest.fixture
-def stack(tmp_path):
-    """Config, library root, and a factory for *independent* session factories —
-    each thread must own its session (concept §16 database rule 1)."""
-    config, lib = _config(tmp_path)
-    engines = []
-
-    def factory():
-        engine = create_db_engine(config.database_url)
-        engines.append(engine)
-        return create_session_factory(engine)
-
-    yield config, lib, factory
-    for engine in engines:
-        engine.dispose()
-
-
-def image(path: Path, seed: int) -> None:
-    path.parent.mkdir(parents=True, exist_ok=True)
-    pixels = np.random.default_rng(seed).integers(0, 256, (64, 96, 3), dtype=np.uint8)
-    Image.fromarray(pixels).save(path, quality=90)
-
-
-def register(sf, path: Path) -> str:
-    asset_id = str(uuid.uuid4())
-    with sf() as session:
-        session.add(
-            Asset(
-                id=asset_id,
-                original_path=str(path),
-                current_path=str(path),
-                discovered_at=NOW,
-                hash_version=1,
-                byte_size=path.stat().st_size,
-                current_sha256=hashing.sha256_file(path),
-            )
-        )
-        session.commit()
-    return asset_id
-
-
-def album(sf, lib: Path, name: str, *, approved_name: str, count: int = 2) -> list[str]:
-    """A real folder of real images with an approved proposal renaming it."""
-    folder = lib / name
-    ids = []
-    for index in range(count):
-        path = folder / f"{name}-{index}.jpg"
-        image(path, seed=abs(hash((name, index))) % 10_000)
-        ids.append(register(sf, path))
-    with sf() as session:
-        session.add(
-            AlbumProposal(
-                id=str(uuid.uuid4()),
-                album=name,
-                proposed_name=approved_name,
-                final_name=approved_name,
-                status="approved",
-                version=2,
-            )
-        )
-        session.commit()
-    return ids
-
-
-def contents(lib: Path) -> Counter:
-    """Every file body under the library — what may never be lost or duplicated."""
-    return Counter(
-        path.read_bytes()
-        for path in lib.rglob("*")
-        if path.is_file() and ".rename-" not in path.name
-    )
-
-
-def integrity_ok(sf) -> bool:
-    with sf() as session:
-        return session.execute(text("PRAGMA integrity_check")).scalar() == "ok"
-
-
-# ── database writer pressure ─────────────────────────────────────────────────
-
-
-def test_many_writers_finish_without_losing_a_row(stack, rng):
-    """Eight lanes committing at once: SQLite has one writer, so this either works
-    through the busy timeout or loses data. Nothing may be lost."""
-    _, _, factory = stack
-    service = JobService(factory())
-    job_ids = [service.enqueue("scan", items=[f"i{n}" for n in range(4)])["id"] for n in range(8)]
-
-    def write(index: int) -> None:
-        own = JobService(factory())
-        job_id = job_ids[index]
-        token = own.claim(["scan"], f"w{index}")
-        if token is None:
-            return
-        for item in [f"i{n}" for n in range(4)]:
-            jitter(rng)
-            own.set_item(token["id"], item, ItemState.RUNNING, fencing_token=token["fencing_token"])
-            own.set_item(
-                token["id"], item, ItemState.SUCCEEDED, fencing_token=token["fencing_token"]
-            )
-        assert job_id  # the claim order is racy; every job is claimed by someone
-
-    with ThreadPoolExecutor(max_workers=8) as pool:
-        list(pool.map(write, range(8)))
-
-    for job_id in job_ids:
-        progress = service.progress(job_id)
-        assert progress["by_state"].get(ItemState.SUCCEEDED) == 4, progress
-    assert integrity_ok(factory())
-
-
-# ── worker claim races ───────────────────────────────────────────────────────
-
-
-def test_a_job_is_claimed_by_exactly_one_worker(stack, rng):
-    _, _, factory = stack
-    service = JobService(factory())
-    job_ids = {service.enqueue("scan", items=["a"])["id"] for _ in range(10)}
-
-    claimed: list[str] = []
-    lock = threading.Lock()
-
-    def claim_all(index: int) -> None:
-        own = JobService(factory())
-        while True:
-            jitter(rng)
-            job = own.claim(["scan"], f"w{index}")
-            if job is None:
-                return
-            with lock:
-                claimed.append(job["id"])
-
-    with ThreadPoolExecutor(max_workers=4) as pool:
-        list(pool.map(claim_all, range(4)))
-
-    assert sorted(claimed) == sorted(job_ids), "every job claimed exactly once"
-    assert len(set(claimed)) == len(claimed)
-
-
-def test_a_late_commit_from_an_expired_lease_is_refused(stack, rng):
-    """Lease expiry then a late write from the old owner: the fencing token, not
-    timing, decides who may commit."""
-    _, _, factory = stack
-    service = JobService(factory())
-    job = service.enqueue("scan", items=["a"])
-    old = service.claim(["scan"], "old")
-    service.recover_stale(now=_now() + timedelta(hours=1))
-    new = service.claim(["scan"], "new")
-
-    jitter(rng)
-    with pytest.raises(JobConflict):
-        service.set_item(job["id"], "a", ItemState.RUNNING, fencing_token=old["fencing_token"])
-    with pytest.raises(JobConflict):
-        service.transition(job["id"], JobState.SUCCEEDED, fencing_token=old["fencing_token"])
-
-    # The current owner finishes the same work without interference.
-    service.set_item(job["id"], "a", ItemState.RUNNING, fencing_token=new["fencing_token"])
-    service.set_item(job["id"], "a", ItemState.SUCCEEDED, fencing_token=new["fencing_token"])
-    service.transition(job["id"], JobState.SUCCEEDED, fencing_token=new["fencing_token"])
-    assert service.get(job["id"])["state"] == JobState.SUCCEEDED
-    assert integrity_ok(factory())
-
-
-def test_cancellation_arriving_at_a_random_moment_leaves_a_consistent_job(stack, rng):
-    _, _, factory = stack
-    service = JobService(factory())
-    job = service.enqueue("scan", items=[f"i{n}" for n in range(6)])
-    handled: list[str] = []
-
-    def handler(item, ctx):
-        jitter(rng)
-        handled.append(item)
-
-    def cancel() -> None:
-        try:
-            service.cancel(job["id"])
-        except (JobConflict, InvalidTransition):
-            # The request lost its race with the worker's own transition; the job
-            # simply finishes. What must never happen is a *silent* overwrite.
-            pass
-
-    canceller = threading.Timer(rng.uniform(0, 0.03), cancel)
-    canceller.start()
-    Worker(factory(), {"scan": handler}, "w1").run_once()
-    canceller.join()
-
-    state = service.get(job["id"])["state"]
-    assert state in {JobState.SUCCEEDED, JobState.CANCELLED, JobState.CANCELLING}
-    by_state = service.progress(job["id"])["by_state"]
-    assert by_state.get(ItemState.RUNNING, 0) == 0, "no item left mid-flight"
-    assert len(handled) == len(set(handled)), "no item ran twice"
-    if state == JobState.CANCELLED:
-        # Whatever did not run is resumable, never silently dropped.
-        assert by_state.get(ItemState.SUCCEEDED, 0) + by_state.get(ItemState.QUEUED, 0) == 6
-    assert integrity_ok(factory())
-
-
-KILLABLE_WORKER = """
-import sys
-sys.path.insert(0, {repo!r})
-from pathlib import Path
-from photo_pipeline.db import create_db_engine, create_session_factory
-from photo_pipeline.jobs.worker import Worker
-
-db_url, log = sys.argv[1], Path(sys.argv[2])
-
-def handler(item_key, ctx):
-    with log.open("a") as handle:
-        handle.write(item_key + "\\n")
-
-sf = create_session_factory(create_db_engine(db_url))
-Worker(sf, {{"scan": handler}}, sys.argv[3], lease_seconds=1).run_once()
-"""
-
-
-def test_a_worker_killed_at_a_random_item_resumes_exactly_once(stack, rng, tmp_path):
-    """Kill a real worker mid-batch, restart it, and assert the invariant across
-    the restart: every item ran exactly once and the job ends succeeded."""
-    config, _, factory = stack
-    service = JobService(factory())
-    items = [f"i{n}" for n in range(rng.randint(2, 5))]
-    job = service.enqueue("scan", items=items)
-    log = tmp_path / "handled.log"
-    script = tmp_path / "killable_worker.py"
-    script.write_text(KILLABLE_WORKER.format(repo=str(REPO)))
-
-    def run_worker(worker_id: str, barrier: str | None) -> None:
-        env = dict(os.environ)
-        env.pop("PHOTO_PIPELINE_FAULT_AFTER", None)
-        if barrier:
-            env["PHOTO_PIPELINE_FAULT_AFTER"] = barrier
-        subprocess.run(
-            [sys.executable, str(script), config.database_url, str(log), worker_id],
-            env=env,
-            capture_output=True,
-        )
-
-    run_worker("killable", JOB_ITEM_DONE)  # dies after its first completed item
-    assert log.read_text().split() == items[:1]
-    time.sleep(1.1)  # let the dead worker's one-second lease expire
-    jitter(rng)
-    run_worker("survivor", None)
-
-    assert service.get(job["id"])["state"] == JobState.SUCCEEDED
-    handled = log.read_text().split()
-    assert sorted(handled) == sorted(items), f"an item ran twice or not at all: {handled}"
-    assert integrity_ok(factory())
-
-
-# ── filesystem races ─────────────────────────────────────────────────────────
-
-
-def test_a_file_changed_during_apply_never_loses_content(stack, rng):
-    """The user edits a photo while its folder is being renamed. The rename may
-    win or be refused, but no body may disappear or be overwritten."""
-    config, lib, factory = stack
-    sf = factory()
-    album(sf, lib, "rome", approved_name="2019 Rome")
-    plan = RenameService(sf, library_roots=(lib,)).build_plan()
-    before = contents(lib)
-
-    edited = b"the user saved over this file"
-
-    def edit() -> None:
-        jitter(rng)
-        target = next((lib / "rome").glob("*.jpg"), None)
-        if target is None:
-            return  # the rename won the race; the folder already moved
-        try:
-            target.write_bytes(edited)
-        except OSError:
-            pass  # ...or it moved between the glob and the write
-
-    thread = threading.Thread(target=edit)
-    thread.start()
-    RenameApplyService(sf, library_roots=(lib,)).apply(plan["id"], expected_version=plan["version"])
-    thread.join()
-
-    after = contents(lib)
-    assert sum(after.values()) == sum(before.values()), "a file was lost or duplicated"
-    survived = (before - Counter({edited: 1})) & after
-    assert sum(survived.values()) >= sum(before.values()) - 1, "unrelated content was destroyed"
-    assert integrity_ok(sf)
-
-
-def test_two_folders_claiming_one_destination_never_merge(stack, rng):
-    """Two approved albums want the same name. The plan must refuse rather than
-    move one folder into the other."""
-    config, lib, factory = stack
-    sf = factory()
-    album(sf, lib, "rome-a", approved_name="2019 Rome")
-    album(sf, lib, "rome-b", approved_name="2019 Rome")
-    jitter(rng)
-
-    plan = RenameService(sf, library_roots=(lib,)).build_plan()
-    codes = {issue["code"] for op in plan["operations"] for issue in op["issues"]}
-
-    assert "duplicate_target" in codes
-    assert plan["applicable"] is False
-    assert (lib / "rome-a").is_dir() and (lib / "rome-b").is_dir()
-
-
-def test_thumbnail_requests_racing_a_rename_never_serve_a_wrong_file(stack, rng):
-    """Previews are keyed by pixels, not paths, so a rename must not make a request
-    fail loudly *or* return another asset's picture."""
-    config, lib, factory = stack
-    sf = factory()
-    asset_ids = album(sf, lib, "rome", approved_name="2019 Rome")
-    plan = RenameService(sf, library_roots=(lib,)).build_plan()
-
-    thumbnails = ThumbnailService(factory(), config)
-    expected = {aid: thumbnails.generate(aid, 256).read_bytes() for aid in asset_ids}
-    served: dict[str, set[bytes]] = {aid: set() for aid in asset_ids}
-    errors: list[str] = []
-    stop = threading.Event()
-
-    def serve() -> None:
-        own = ThumbnailService(factory(), config)
-        while not stop.is_set():
-            for asset_id in asset_ids:
-                jitter(rng)
-                try:
-                    served[asset_id].add(own.generate(asset_id, 256).read_bytes())
-                except ThumbnailError as error:
-                    errors.append(error.code)  # precise, never an unhandled crash
-
-    reader = threading.Thread(target=serve)
-    reader.start()
-    try:
-        RenameApplyService(sf, library_roots=(lib,)).apply(
-            plan["id"], expected_version=plan["version"]
-        )
-    finally:
-        stop.set()
-        reader.join()
-
-    for asset_id, bodies in served.items():
-        assert bodies <= {expected[asset_id]}, "a request served another asset's picture"
-    assert integrity_ok(sf)
-
-
-# ── stage races ──────────────────────────────────────────────────────────────
-
-
-def test_an_analysis_result_racing_a_safety_flip_is_discarded(stack, rng):
-    """Concept §18 scenario 7: the reviewer marks an asset NSFW while the provider
-    call is in flight. The answer that comes back describes an asset that may no
-    longer be analysed, so it is dropped — and no analysis EXIF is written."""
-    _, lib, factory = stack
-    sf = factory()
-    path = lib / "beach.jpg"
-    image(path, seed=7)
-    asset_id = register(sf, path)
-    with sf() as session:
-        session.add(
-            SafetyReview(id=str(uuid.uuid4()), asset_id=asset_id, decision="sfw", created_at=NOW)
-        )
-        session.commit()
-    before_sha = hashing.sha256_file(path)
-
-    class FlippingProvider:
-        """Records the call, then the reviewer's decision lands mid-flight."""
-
-        def __init__(self) -> None:
-            self.calls: list[str] = []
-
-        def analyze(self, path, *, album_hint):
-            self.calls.append(path)
-            jitter(rng)
-            with sf() as session:
-                session.add(
-                    SafetyReview(
-                        id=str(uuid.uuid4()),
-                        asset_id=asset_id,
-                        decision="nsfw",
-                        prior_decision="sfw",
-                        created_at=NOW + timedelta(minutes=1),
-                    )
-                )
-                session.commit()
-            return {"description": "a beach", "tags": ["beach", "sand"]}
-
-    provider = FlippingProvider()
-    result = AnalysisService(sf, provider=provider, library_roots=(lib,)).run([asset_id])
-
-    assert provider.calls == [str(path)], "the call was legitimate when it started"
-    assert result == {"analyzed": 0, "skipped": 1, "errors": 0}
-    with sf() as session:
-        row = session.get(AnalysisResult, asset_id)
-        assert row.status == "skipped_nsfw" and row.description is None
-        assert session.get(ExifProjection, (asset_id, "analysis")) is None
-    assert hashing.sha256_file(path) == before_sha, "the file was written after the flip"
-
-
-def test_a_safety_decision_taken_twice_at_once_keeps_one_history(stack, rng):
-    """Two windows, one asset, the same decision: the audit trail may record both
-    attempts, but the effective state must be a single coherent decision."""
-    _, lib, factory = stack
-    sf = factory()
-    path = lib / "twice.jpg"
-    image(path, seed=11)
-    asset_id = register(sf, path)
-
-    from photo_pipeline.services.safety import SafetyService
-
-    def decide(decision: str) -> None:
-        jitter(rng)
-        SafetyService(factory()).decide(asset_id, decision, write_exif=False)
-
-    with ThreadPoolExecutor(max_workers=2) as pool:
-        list(pool.map(decide, ["sfw", "sfw"]))
-
-    with sf() as session:
-        decisions = [
-            review.decision
-            for review in session.scalars(
-                select(SafetyReview)
-                .where(SafetyReview.asset_id == asset_id)
-                .order_by(SafetyReview.created_at)
-            )
-        ]
-    assert decisions and set(decisions) == {"sfw"}
-    assert integrity_ok(sf)

tests/integration/test_exif_checkpoints.py

@@ -1,297 +0,0 @@
-"""US07-03: EXIF checkpoints, asserted with before/after metadata snapshots.
-
-Every test here reads the complete metadata of a real file through exiftool before
-the stage runs and again afterwards, then compares the two. That is the only way to
-prove the property the concept actually asks for: a stage owns a few fields and must
-leave literally everything else — dates, GPS, camera, artist, rating, other people's
-keywords — exactly as it found them.
-
-The other half is divergence. When something outside the stage's ownership does move,
-the checkpoint must say so, refuse to call itself verified, and change nothing back:
-a silent repair is how a library quietly loses the user's metadata.
-"""
-
-from __future__ import annotations
-
-import json
-import shutil
-import subprocess
-import uuid
-from datetime import datetime, timezone
-from types import SimpleNamespace
-
-import pytest
-
-from photo_pipeline.config import Config
-from photo_pipeline.db import create_db_engine, create_session_factory, run_migrations
-from photo_pipeline.integrations import exiftool
-from photo_pipeline.models import Asset, ExifProjection
-from photo_pipeline.services import exif_checkpoint
-from photo_pipeline.services.analysis import AnalysisService
-from photo_pipeline.services.safety import SafetyService
-from tests.fixtures.media_corpus import CASES_BY_ID, build_corpus
-
-pytestmark = pytest.mark.skipif(
-    shutil.which("exiftool") is None, reason="exiftool not installed"
-)
-
-
-@pytest.fixture
-def env(tmp_path):
-    data = tmp_path / "data"
-    data.mkdir()
-    lib = tmp_path / "lib"
-    files = build_corpus(
-        lib,
-        ids=(
-            "user_exif",
-            "prior_safety_keyword",
-            "prior_analysis_keywords",
-            "conflicting_safety_keywords",
-            "malformed_metadata",
-            "no_exif",
-        ),
-    )
-    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)
-    assets = {}
-    with sf() as session:
-        for case_id, path in files.items():
-            asset = Asset(
-                id=str(uuid.uuid4()),
-                original_path=str(path),
-                current_path=str(path),
-                discovered_at=datetime.now(timezone.utc),
-                hash_version=1,
-            )
-            session.add(asset)
-            assets[case_id] = asset.id
-        session.commit()
-    yield SimpleNamespace(config=config, lib=lib, sf=sf, files=files, assets=assets)
-    engine.dispose()
-
-
-class Provider:
-    """A vision provider whose tags are fixed, so the EXIF assertion is the test."""
-
-    def __init__(self, tags):
-        self.tags = list(tags)
-
-    def analyze(self, path, *, album_hint):
-        return {"description": "a photo", "tags": self.tags}
-
-
-def snapshot(path):
-    return exiftool.read_all(str(path))
-
-
-def stage_fields(before, after):
-    """Everything that changed except this application's owned and volatile tags."""
-    return exif_checkpoint.compare(before, after)
-
-
-# ── preservation ──────────────────────────────────────────────────────────────
-
-
-def test_a_safety_decision_preserves_every_user_field(env):
-    case = CASES_BY_ID["user_exif"]
-    path = env.files["user_exif"]
-    before = snapshot(path)
-    assert before, "the fixture must actually carry user metadata"
-
-    result = SafetyService(env.sf).decide(env.assets["user_exif"], "nsfw")
-    after = snapshot(path)
-
-    assert result["exif_verified"] is True
-    assert stage_fields(before, after) == ()
-    for field in case.preserved_fields:
-        assert after[field] == before[field], field
-    assert "nsfw" in exif_checkpoint.owned_values(after)
-    assert "sfw" not in exif_checkpoint.owned_values(after)
-
-
-def test_analysis_keywords_are_additive_and_keep_the_safety_decision(env):
-    """The two stages share the Keywords field; the second must merge, not replace."""
-    asset_id = env.assets["prior_analysis_keywords"]
-    path = env.files["prior_analysis_keywords"]
-    SafetyService(env.sf).decide(asset_id, "sfw")
-    before = snapshot(path)
-
-    AnalysisService(
-        env.sf, provider=Provider(["harbour", "boats"]), library_roots=(env.lib,)
-    ).run([asset_id])
-    after = snapshot(path)
-
-    keywords = exif_checkpoint.owned_values(after)
-    assert {"sfw", "beach", "sunset", "harbour", "boats"} <= keywords
-    assert "nsfw" not in keywords
-    assert stage_fields(before, after) == ()
-
-
-def test_flipping_a_safety_decision_removes_only_the_opposite_keyword(env):
-    asset_id = env.assets["prior_analysis_keywords"]
-    path = env.files["prior_analysis_keywords"]
-    safety = SafetyService(env.sf)
-    safety.decide(asset_id, "nsfw")
-    before = snapshot(path)
-
-    safety.decide(asset_id, "sfw")
-    after = snapshot(path)
-
-    keywords = exif_checkpoint.owned_values(after)
-    assert "sfw" in keywords and "nsfw" not in keywords
-    assert {"beach", "sunset"} <= keywords, "analysis keywords are not safety's to remove"
-    assert stage_fields(before, after) == ()
-
-
-def test_conflicting_safety_keywords_are_resolved_to_one(env):
-    asset_id = env.assets["conflicting_safety_keywords"]
-    SafetyService(env.sf).decide(asset_id, "sfw")
-    keywords = exif_checkpoint.owned_values(snapshot(env.files["conflicting_safety_keywords"]))
-    assert keywords & {"sfw", "nsfw"} == {"sfw"}
-
-
-def test_malformed_metadata_does_not_block_the_checkpoint(env):
-    result = SafetyService(env.sf).decide(env.assets["malformed_metadata"], "nsfw")
-    assert result["exif_verified"] is True
-
-
-# ── verification and the refreshed hash ───────────────────────────────────────
-
-
-def test_a_verified_checkpoint_refreshes_the_recorded_bytes(env):
-    """exiftool rewrites the container, so the stored SHA-256 must be the new one —
-    upload compares against exactly these bytes."""
-    asset_id = env.assets["no_exif"]
-    path = env.files["no_exif"]
-    SafetyService(env.sf).decide(asset_id, "sfw")
-
-    with env.sf() as session:
-        asset = session.get(Asset, asset_id)
-        row = session.get(ExifProjection, (asset_id, "safety"))
-    import hashlib
-
-    on_disk = hashlib.sha256(path.read_bytes()).hexdigest()
-    assert asset.current_sha256 == on_disk
-    assert row.state == "verified" and row.result_file_sha256 == on_disk
-    assert json.loads(row.desired_json) == {"add": ["sfw"], "remove": ["nsfw"]}
-
-
-def test_the_projection_survives_a_restart(env):
-    asset_id = env.assets["user_exif"]
-    SafetyService(env.sf).decide(asset_id, "nsfw")
-
-    engine = create_db_engine(env.config.database_url)  # a fresh connection, as a restart is
-    try:
-        with create_session_factory(engine)() as session:
-            row = session.get(ExifProjection, (asset_id, "safety"))
-            assert row.state == "verified" and row.verified_at is not None
-    finally:
-        engine.dispose()
-
-
-# ── divergence ────────────────────────────────────────────────────────────────
-
-
-def test_a_field_the_stage_does_not_own_changing_is_divergent(env, monkeypatch):
-    """Something rewrote the artist while the safety keyword was being written.
-
-    The stage must not call that verified, must record what moved, and must not put
-    the old value back — the file is now a question for a human.
-    """
-    asset_id = env.assets["user_exif"]
-    path = env.files["user_exif"]
-    real_apply = exiftool.apply_keywords
-
-    def sabotage(target, *, add=(), remove=()):
-        ok = real_apply(target, add=add, remove=remove)
-        subprocess.run(
-            ["exiftool", "-m", "-overwrite_original", "-Artist=Someone Else", str(target)],
-            capture_output=True,
-            check=False,
-        )
-        return ok
-
-    monkeypatch.setattr(exiftool, "apply_keywords", sabotage)
-    result = SafetyService(env.sf).decide(asset_id, "nsfw")
-
-    assert result["exif_verified"] is False, "a divergent checkpoint is not verified"
-    with env.sf() as session:
-        row = session.get(ExifProjection, (asset_id, "safety"))
-        asset = session.get(Asset, asset_id)
-    assert row.state == "divergent"
-    assert "EXIF:IFD0:Artist" in json.loads(row.divergent_fields)
-    assert row.verified_at is None
-    # Not repaired, and not silently accepted as the current verified bytes.
-    assert snapshot(path)["EXIF:IFD0:Artist"] == "Someone Else"
-    assert asset.current_sha256 is None
-
-
-def test_a_divergent_asset_shows_up_in_the_review_queue(env, monkeypatch):
-    asset_id = env.assets["user_exif"]
-    monkeypatch.setattr(
-        exif_checkpoint,
-        "run",
-        lambda *args, **kwargs: exif_checkpoint.CheckpointResult(
-            exif_checkpoint.DIVERGENT, changed_fields=("EXIF:IFD0:Artist",), sha256="abc"
-        ),
-    )
-    SafetyService(env.sf).decide(asset_id, "sfw")
-
-    rows = SafetyService(env.sf).review_queue()["items"]
-    row = next(item for item in rows if item["asset_id"] == asset_id)
-    assert row["exif_state"] == "divergent"
-    assert row["exif_verified"] is False
-
-
-def test_a_write_that_does_not_take_is_a_failure_not_a_verification(env, monkeypatch):
-    monkeypatch.setattr(exiftool, "apply_keywords", lambda *a, **k: False)
-    result = SafetyService(env.sf).decide(env.assets["no_exif"], "sfw")
-    assert result["exif_verified"] is False
-    assert exif_checkpoint.state_for(env.sf, env.assets["no_exif"], "safety") == "failed"
-
-
-def test_unreadable_metadata_is_a_failure_not_an_empty_snapshot(env, monkeypatch):
-    """``None`` from exiftool means "cannot answer"; treating it as "nothing there"
-    would make every field look preserved."""
-    monkeypatch.setattr(exiftool, "read_all", lambda path: None)
-    result = exif_checkpoint.run(str(env.files["no_exif"]), add=("sfw",))
-    assert result.state == "failed" and result.reason == "metadata_unreadable"
-
-
-def test_analysis_records_its_own_projection_separately(env):
-    asset_id = env.assets["user_exif"]
-    SafetyService(env.sf).decide(asset_id, "sfw")
-    AnalysisService(env.sf, provider=Provider(["pier"]), library_roots=(env.lib,)).run([asset_id])
-
-    assert exif_checkpoint.state_for(env.sf, asset_id, "safety") == "verified"
-    assert exif_checkpoint.state_for(env.sf, asset_id, "analysis") == "verified"
-
-
-# ── the comparison rules themselves ───────────────────────────────────────────
-
-
-def test_compare_ignores_owned_and_volatile_fields_only():
-    before = {
-        "EXIF:IFD0:Artist": "Ada",
-        "IPTC:Keywords": ["holiday"],
-        "File:System:FileSize": "3.8 kB",
-        "File:CurrentIPTCDigest": "aaa",
-        "XMP:XMP-x:XMPToolkit": "old",
-    }
-    after = {
-        "EXIF:IFD0:Artist": "Ada",
-        "IPTC:Keywords": ["holiday", "sfw"],
-        "File:System:FileSize": "3.9 kB",
-        "File:CurrentIPTCDigest": "bbb",
-        "XMP:XMP-x:XMPToolkit": "new",
-    }
-    assert exif_checkpoint.compare(before, after) == ()
-
-    after["EXIF:GPS:GPSLatitude"] = "48.1"  # an addition counts as much as a loss
-    del after["EXIF:IFD0:Artist"]
-    assert exif_checkpoint.compare(before, after) == ("EXIF:GPS:GPSLatitude", "EXIF:IFD0:Artist")

tests/integration/test_fault_matrix.py

@@ -1,527 +0,0 @@
-"""The fault matrix (US07-04, concept §18 "crash/fault-injection tests").
-
-Process death at each persisted transition lives in tests/e2e/test_crash_recovery.py
-and the per-stage recovery suites. This file covers the *environmental* faults —
-the ones that are not a crash but are just as good at corrupting a library if the
-code guesses:
-
-    disk full · read-only path · database busy · database corruption ·
-    network failure · malformed provider output · GPU exhaustion ·
-    subprocess hang · missing external tool
-
-Every case asserts the same shape of outcome: the operation fails visibly, the
-failure names what happened, and nothing irreversible was done on the way — no
-source removed, no metadata marked verified, no decision invented.
-"""
-
-from __future__ import annotations
-
-import errno
-import os
-import sqlite3
-import stat
-import threading
-import time
-import uuid
-from contextlib import contextmanager
-from datetime import datetime, timezone
-from pathlib import Path
-
-import numpy as np
-import pytest
-from PIL import Image
-from sqlalchemy import select, text
-from sqlalchemy.exc import DatabaseError, OperationalError
-
-from photo_pipeline import faults
-from photo_pipeline.config import Config
-from photo_pipeline.db import create_db_engine, create_session_factory, run_migrations
-from photo_pipeline.jobs.worker import Worker
-from photo_pipeline.models import (
-    AlbumProposal,
-    AnalysisResult,
-    Asset,
-    ExifProjection,
-    SafetyReview,
-    UploadBatch,
-    UploadItem,
-)
-from photo_pipeline.services import archive_transfer, exif_checkpoint, hashing, rename_apply
-from photo_pipeline.services.analysis import AnalysisService
-from photo_pipeline.services.archive_transfer import ArchiveTransferService
-from photo_pipeline.services.archives import ArchiveService
-from photo_pipeline.services.jobs import ItemState, JobService
-from photo_pipeline.services.rename_apply import ApplyError, RenameApplyService
-from photo_pipeline.services.rename_journal import JournalState, RenameJournal
-from photo_pipeline.services.renames import RenameService
-from photo_pipeline.services.safety import SafetyService
-from photo_pipeline.services.uploads import UploadService
-
-NOW = datetime(2026, 1, 1, tzinfo=timezone.utc)
-
-
-# ── environment ──────────────────────────────────────────────────────────────
-
-
-def _env(tmp_path, **extra):
-    (tmp_path / "data").mkdir(exist_ok=True)
-    lib = tmp_path / "lib"
-    lib.mkdir(exist_ok=True)
-    config = Config.from_env(
-        {
-            "PHOTO_PIPELINE_DATA_DIR": str(tmp_path / "data"),
-            "PHOTO_PIPELINE_LIBRARY_ROOTS": str(lib),
-            "PHOTO_PIPELINE_ARCHIVE_FREE_SPACE_RESERVE_BYTES": "0",
-            **extra,
-        }
-    )
-    run_migrations(config.database_url)
-    return config, create_session_factory(create_db_engine(config.database_url)), lib
-
-
-def image(path: Path, seed: int = 1) -> None:
-    path.parent.mkdir(parents=True, exist_ok=True)
-    pixels = np.random.default_rng(seed).integers(0, 256, (64, 96, 3), dtype=np.uint8)
-    Image.fromarray(pixels).save(path, quality=90)
-
-
-def register(sf, path: Path) -> str:
-    asset_id = str(uuid.uuid4())
-    with sf() as session:
-        session.add(
-            Asset(
-                id=asset_id,
-                original_path=str(path),
-                current_path=str(path),
-                discovered_at=NOW,
-                hash_version=1,
-                byte_size=path.stat().st_size,
-                current_sha256=hashing.sha256_file(path),
-            )
-        )
-        session.commit()
-    return asset_id
-
-
-def uploaded_album(sf, lib, album="rome", names=("a.jpg",)):
-    """An album with verified upload evidence — what archiving requires."""
-    folder = lib / album
-    folder.mkdir(parents=True, exist_ok=True)
-    with sf() as session:
-        batch_id = str(uuid.uuid4())
-        session.add(
-            UploadBatch(
-                id=batch_id,
-                album=album,
-                folder=str(folder),
-                album_name=album,
-                state="succeeded",
-                preflight_token="v1:test",
-                outcome_state="verified",
-                created_at=NOW,
-            )
-        )
-        for name in names:
-            path = folder / name
-            path.write_bytes(f"{album}/{name} content".encode() * 8)
-            asset_id = str(uuid.uuid4())
-            session.add(
-                Asset(
-                    id=asset_id,
-                    original_path=str(path),
-                    current_path=str(path),
-                    discovered_at=NOW,
-                    hash_version=1,
-                    byte_size=path.stat().st_size,
-                    current_sha256=hashing.sha256_file(path),
-                )
-            )
-            session.add(
-                UploadItem(
-                    batch_id=batch_id,
-                    asset_id=asset_id,
-                    path=str(path),
-                    sha256=hashing.sha256_file(path),
-                    sha1="0" * 40,
-                    state="sent",
-                    outcome="uploaded",
-                )
-            )
-        session.commit()
-    return folder
-
-
-def archive_plan(sf, config, archive, albums=None):
-    location = ArchiveService(sf, config=config).register("external", str(archive))
-    token = ArchiveService(sf, config=config).preflight(location["id"], albums)["token"]
-    service = ArchiveTransferService(sf, config=config)
-    return service, service.create(location["id"], albums, token=token)
-
-
-def fake_tool(directory: Path, name: str, body: str) -> Path:
-    """A real executable on a directory a test can put in front of PATH."""
-    directory.mkdir(parents=True, exist_ok=True)
-    path = directory / name
-    path.write_text(f"#!/bin/sh\n{body}")
-    path.chmod(path.stat().st_mode | stat.S_IEXEC | stat.S_IXGRP | stat.S_IXOTH)
-    return path
-
-
-# ── control points ───────────────────────────────────────────────────────────
-
-
-def test_the_fault_barrier_does_nothing_unless_its_variable_names_the_point(monkeypatch):
-    monkeypatch.delenv(faults.ENV_VAR, raising=False)
-    for point in (faults.EXIF_WRITTEN, faults.UPLOAD_ACCEPTED, faults.JOB_ITEM_DONE, "moving"):
-        faults.maybe_fault(point)  # would kill the process if it were armed
-    monkeypatch.setenv(faults.ENV_VAR, faults.EXIF_WRITTEN)
-    faults.maybe_fault(faults.UPLOAD_ACCEPTED)  # a different point stays inert
-
-
-def test_no_route_or_configuration_can_arm_a_fault():
-    """The control points are reachable only through an environment variable read
-    inside ``photo_pipeline.faults`` — never through the API, and never through
-    configuration a browser or a config file could set."""
-    from photo_pipeline.api.app import create_app
-
-    app = create_app()
-    assert not [route for route in app.routes if "fault" in getattr(route, "path", "")]
-    assert not [field for field in Config.model_fields if "fault" in field]
-
-    package = Path(__file__).resolve().parents[2] / "photo_pipeline"
-    sources = {
-        path.relative_to(package.parent)
-        for path in package.rglob("*.py")
-        if faults.ENV_VAR in path.read_text()
-    }
-    assert sources == {Path("photo_pipeline/faults.py")}
-
-
-# ── disk full ────────────────────────────────────────────────────────────────
-
-
-def test_a_full_disk_during_an_archive_never_removes_the_source(tmp_path, monkeypatch):
-    config, sf, lib = _env(tmp_path)
-    archive = tmp_path / "archive"
-    archive.mkdir()
-    folder = uploaded_album(sf, lib)
-    original = {path: path.read_bytes() for path in folder.iterdir()}
-    service, plan = archive_plan(sf, config, archive)
-
-    def no_space(*args, **kwargs):
-        raise OSError(errno.ENOSPC, "No space left on device")
-
-    # Force the cross-filesystem path (a real archive medium) and fill it up.
-    monkeypatch.setattr(archive_transfer, "_same_filesystem", lambda *a: False)
-    monkeypatch.setattr(archive_transfer.shutil, "copyfileobj", no_space)
-
-    result = service.apply(plan["id"])
-
-    assert result["archived"] == 0 and result["failed"] == 1
-    for path, body in original.items():
-        assert path.read_bytes() == body, "the source was touched despite the failure"
-    with sf() as session:
-        assert all(a.availability_state == "active" for a in session.scalars(select(Asset)))
-    assert [p for p in archive.rglob("*") if p.is_file() and not p.name.startswith(".")] == []
-
-
-# ── read-only paths ──────────────────────────────────────────────────────────
-
-
-@pytest.mark.skipif(os.geteuid() == 0, reason="root ignores directory permissions")
-def test_a_read_only_library_refuses_the_rename_and_keeps_the_source(tmp_path):
-    config, sf, lib = _env(tmp_path)
-    folder = lib / "rome"
-    image(folder / "a.jpg")
-    register(sf, folder / "a.jpg")
-    with sf() as session:
-        session.add(
-            AlbumProposal(
-                id=str(uuid.uuid4()),
-                album="rome",
-                proposed_name="2019 Rome",
-                final_name="2019 Rome",
-                status="approved",
-                version=2,
-            )
-        )
-        session.commit()
-    plan = RenameService(sf, library_roots=(lib,)).build_plan()
-
-    mode = lib.stat().st_mode
-    lib.chmod(0o500)  # readable, traversable, not writable
-    try:
-        result = RenameApplyService(sf, library_roots=(lib,)).apply(
-            plan["id"], expected_version=plan["version"]
-        )
-    finally:
-        lib.chmod(mode)
-
-    assert result["applied"] == 0 and result["failed"] == 1
-    assert (folder / "a.jpg").exists() and not (lib / "2019 Rome").exists()
-    with sf() as session:
-        asset = session.scalars(select(Asset)).one()
-        assert asset.current_path == str(folder / "a.jpg")
-
-
-# ── database faults ──────────────────────────────────────────────────────────
-
-
-@contextmanager
-def exclusive_lock(database_url: str):
-    """Hold SQLite's write lock from another thread, the way a second process would.
-
-    A sqlite3 connection belongs to the thread that opened it, so the holder thread
-    opens, locks, waits, and releases entirely on its own.
-    """
-    path = database_url.replace("sqlite:///", "")
-    locked, release = threading.Event(), threading.Event()
-
-    def hold() -> None:
-        connection = sqlite3.connect(path, timeout=10)
-        connection.execute("BEGIN EXCLUSIVE")
-        locked.set()
-        release.wait(30)
-        connection.rollback()
-        connection.close()
-
-    thread = threading.Thread(target=hold)
-    thread.start()
-    assert locked.wait(10), "the holder never acquired the lock"
-    try:
-        yield release.set  # callers may release early; exiting releases anyway
-    finally:
-        release.set()
-        thread.join(10)
-
-
-def test_a_busy_database_waits_rather_than_failing(tmp_path):
-    """SQLite has one writer. A short conflict must resolve through the busy
-    timeout instead of surfacing as an error."""
-    config, sf, lib = _env(tmp_path)
-    service = JobService(sf)
-    job = service.enqueue("scan", items=["a"])
-
-    with exclusive_lock(config.database_url) as release:
-        threading.Timer(0.3, release).start()
-        started = time.monotonic()
-        claimed = service.claim(["scan"], "w1")  # blocks until the lock is gone
-        waited = time.monotonic() - started
-
-    assert claimed["id"] == job["id"] and claimed["state"] == "running"
-    assert waited >= 0.25, "the claim did not actually wait for the writer"
-    with sf() as session:
-        assert session.execute(text("PRAGMA busy_timeout")).scalar() >= 1000
-
-
-def test_a_database_locked_beyond_the_timeout_is_an_error_not_a_silent_skip(tmp_path):
-    config, sf, lib = _env(tmp_path)
-    with sf() as session:
-        session.execute(text("SELECT 1"))  # connect first: the lock comes after
-        with exclusive_lock(config.database_url):
-            session.execute(text("PRAGMA busy_timeout=50"))  # do not wait five seconds
-            with pytest.raises(OperationalError, match="locked"):
-                session.execute(
-                    text("INSERT INTO jobs (id, job_type, state) VALUES ('x','scan','queued')")
-                )
-                session.commit()
-            session.rollback()
-
-    # The refusal left nothing behind, and the database is still sound.
-    with sf() as session:
-        assert session.execute(text("PRAGMA integrity_check")).scalar() == "ok"
-        assert session.execute(text("SELECT count(*) FROM jobs")).scalar() == 0
-
-
-def test_a_corrupt_database_fails_loudly_instead_of_answering_wrongly(tmp_path):
-    config, sf, lib = _env(tmp_path)
-    for index in range(50):  # enough rows to fill several pages
-        JobService(sf).enqueue("scan", items=[f"item-{index}-{n}" for n in range(20)])
-    with sf() as session:
-        session.execute(text("PRAGMA wal_checkpoint(TRUNCATE)"))  # WAL into the file
-        session.commit()
-    source = Path(config.database_url.replace("sqlite:///", ""))
-
-    corrupt = tmp_path / "corrupt.db"
-    body = bytearray(source.read_bytes())
-    body[4096 : 4096 + 2048] = b"\xde\xad\xbe\xef" * 512  # shred pages, keep the header
-    corrupt.write_bytes(bytes(body))
-
-    engine = create_db_engine(f"sqlite:///{corrupt}")
-    factory = create_session_factory(engine)
-    try:
-        with factory() as session:
-            assert session.execute(text("PRAGMA integrity_check")).scalar() != "ok"
-            # Reading the shredded pages must raise, never return half a table.
-            with pytest.raises(DatabaseError):
-                session.execute(text("SELECT * FROM job_items")).all()
-                session.execute(text("SELECT * FROM job_events")).all()
-                session.execute(text("REINDEX")).all()
-    finally:
-        engine.dispose()
-
-
-# ── external services ────────────────────────────────────────────────────────
-
-
-def test_an_unreachable_immich_blocks_upload_instead_of_starting_one(tmp_path):
-    # Port 9 (discard) refuses connections deterministically.
-    config, sf, lib = _env(
-        tmp_path,
-        PHOTO_PIPELINE_IMMICH_SERVER_URL="http://127.0.0.1:9",
-        PHOTO_PIPELINE_IMMICH_API_KEY="sentinel",
-    )
-    report = UploadService(sf, config=config).preflight()
-    codes = {blocker["code"] for blocker in report["blockers"]}
-    assert "server_unreachable" in codes
-    assert report["state"] != "ready"
-
-
-def test_a_missing_uploader_blocks_upload_with_the_binary_named(tmp_path):
-    config, sf, lib = _env(
-        tmp_path,
-        PHOTO_PIPELINE_IMMICH_SERVER_URL="http://127.0.0.1:9",
-        PHOTO_PIPELINE_IMMICH_API_KEY="sentinel",
-        PHOTO_PIPELINE_IMMICH_GO_BINARY=str(tmp_path / "no-such-immich-go"),
-    )
-    report = UploadService(sf, config=config).preflight()
-    assert "immich_go_missing" in {blocker["code"] for blocker in report["blockers"]}
-
-
-def test_a_malformed_provider_answer_is_a_per_asset_error(tmp_path):
-    config, sf, lib = _env(tmp_path)
-    path = lib / "a.jpg"
-    image(path)
-    asset_id = register(sf, path)
-    with sf() as session:
-        session.add(
-            SafetyReview(id=str(uuid.uuid4()), asset_id=asset_id, decision="sfw", created_at=NOW)
-        )
-        session.commit()
-
-    class MalformedProvider:
-        def analyze(self, path, *, album_hint):
-            raise ValueError("Expecting value: line 1 column 1 (char 0)")
-
-    result = AnalysisService(sf, provider=MalformedProvider(), library_roots=(lib,)).run([asset_id])
-
-    assert result == {"analyzed": 0, "skipped": 0, "errors": 1}
-    with sf() as session:
-        row = session.get(AnalysisResult, asset_id)
-        assert row.status == "error" and "Expecting value" in row.error_message
-        assert session.get(ExifProjection, (asset_id, "analysis")) is None
-
-
-def test_gpu_exhaustion_fails_the_item_without_inventing_a_score(tmp_path):
-    config, sf, lib = _env(tmp_path)
-    path = lib / "a.jpg"
-    image(path)
-    asset_id = register(sf, path)
-
-    class ExhaustedModel:
-        def score(self, paths):
-            raise RuntimeError("MPS backend out of memory (MPS allocated: 9.00 GB)")
-
-    service = SafetyService(sf, model=ExhaustedModel())
-    jobs = JobService(sf)
-    job = jobs.enqueue("safety_score", items=[asset_id])
-    Worker(sf, {"safety_score": lambda item, ctx: service.score_assets([item])}, "w1").run_once()
-
-    progress = jobs.progress(job["id"])
-    assert progress["by_state"] == {ItemState.FAILED: 1}
-    with sf() as session:
-        assert session.scalars(select(SafetyReview)).all() == [], "no score was invented"
-
-
-# ── external tools ───────────────────────────────────────────────────────────
-
-
-def test_a_hanging_exiftool_times_out_and_verifies_nothing(tmp_path, monkeypatch):
-    config, sf, lib = _env(tmp_path)
-    path = lib / "a.jpg"
-    image(path)
-    asset_id = register(sf, path)
-    before = hashing.sha256_file(path)
-
-    fake_tool(tmp_path / "bin", "exiftool", "sleep 30\n")
-    monkeypatch.setenv("PATH", f"{tmp_path / 'bin'}:{os.environ['PATH']}")
-    monkeypatch.setenv("PHOTO_PIPELINE_EXIFTOOL_TIMEOUT", "1")
-
-    review = SafetyService(sf).decide(asset_id, "nsfw")
-
-    # The decision is durable; the metadata claim is not made.
-    assert review["decision"] == "nsfw" and review["exif_verified"] is False
-    assert exif_checkpoint.state_for(sf, asset_id, "safety") == exif_checkpoint.FAILED
-    assert hashing.sha256_file(path) == before
-
-
-def test_a_missing_exiftool_is_a_failed_checkpoint_not_a_verified_one(tmp_path, monkeypatch):
-    config, sf, lib = _env(tmp_path)
-    path = lib / "a.jpg"
-    image(path)
-    asset_id = register(sf, path)
-
-    empty_bin = tmp_path / "empty-bin"
-    empty_bin.mkdir()
-    monkeypatch.setenv("PATH", str(empty_bin))  # nothing on PATH at all
-
-    review = SafetyService(sf).decide(asset_id, "sfw")
-
-    assert review["exif_verified"] is False
-    assert exif_checkpoint.state_for(sf, asset_id, "safety") == exif_checkpoint.FAILED
-    with sf() as session:
-        # Upload eligibility depends on a verified checkpoint, so it stays blocked.
-        assert session.scalars(select(SafetyReview)).all()[-1].exif_verified_at is None
-
-
-def test_a_file_edited_during_the_move_is_left_for_a_human(tmp_path, monkeypatch):
-    """The user saves over a photo in the instant between the move and its
-    verification. The move already happened and the database already followed it,
-    so the operation cannot simply be "failed": it becomes ``rollback_required``
-    and blocks further mutation until someone decides (US07-04)."""
-    config, sf, lib = _env(tmp_path)
-    folder = lib / "rome"
-    image(folder / "a.jpg")
-    register(sf, folder / "a.jpg")
-    with sf() as session:
-        session.add(
-            AlbumProposal(
-                id=str(uuid.uuid4()),
-                album="rome",
-                proposed_name="2019 Rome",
-                final_name="2019 Rome",
-                status="approved",
-                version=2,
-            )
-        )
-        session.commit()
-    plan = RenameService(sf, library_roots=(lib,)).build_plan()
-
-    real_rename = rename_apply.os.rename
-
-    def rename_then_edit(source, destination):
-        real_rename(source, destination)
-        for path in Path(destination).glob("*.jpg"):
-            path.write_bytes(b"the user saved over this file")
-
-    monkeypatch.setattr(rename_apply.os, "rename", rename_then_edit)
-
-    result = RenameApplyService(sf, library_roots=(lib,)).apply(
-        plan["id"], expected_version=plan["version"]
-    )
-
-    assert result["applied"] == 0 and result["failed"] == 1
-    journal = RenameJournal(sf)
-    operation = journal.incomplete()[0]
-    assert operation["journal_state"] == JournalState.ROLLBACK_REQUIRED
-    assert operation["error_code"] == "verify_bytes"
-    assert journal.blocks_mutation() is True, "the unresolved rename blocks the library"
-
-    # Recovery offers the rollback the evidence supports, and the rollback itself
-    # refuses the edited bytes rather than putting the user's newer file back as if
-    # it were the old one.
-    service = RenameApplyService(sf, library_roots=(lib,))
-    with pytest.raises(ApplyError, match="manual recovery"):
-        service.rollback_operation(operation["id"])
-    # Nothing was lost: the edited file is at its new home, not deleted.
-    assert (lib / "2019 Rome" / "a.jpg").read_bytes() == b"the user saved over this file"

tests/integration/test_media_hardening.py

@@ -1,335 +0,0 @@
-"""US07-03: the golden media corpus, driven through the real decode path.
-
-Every case in ``tests/fixtures/media_corpus.py`` is exercised here. The claims:
-
-* a supported format/orientation/profile renders, with the orientation applied and
-  the decode bounded to roughly the size that was asked for;
-* a damaged, empty, lying, or gigapixel file becomes one precise item error — the
-  scan still finishes, the other assets still render, and a job that meets one keeps
-  running;
-* a failure leaves no temporary file behind, and startup removes only the temporaries
-  this service recognises;
-* cache invalidation follows the pixels, not the metadata.
-"""
-
-from __future__ import annotations
-
-import hashlib
-import shutil
-import time
-import tracemalloc
-from types import SimpleNamespace
-
-import pytest
-from PIL import Image
-
-from photo_pipeline import imaging
-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 (
-    TEMP_SUFFIX,
-    ThumbnailError,
-    ThumbnailService,
-)
-from tests.fixtures.media_corpus import CASES, CASES_BY_ID, build_corpus
-
-RENDERABLE = [case for case in CASES if case.expect_error is None]
-DAMAGED = [case for case in CASES if case.expect_error is not None]
-
-needs_exiftool = pytest.mark.skipif(
-    shutil.which("exiftool") is None, reason="exiftool not installed"
-)
-
-
-@pytest.fixture(scope="module")
-def corpus(tmp_path_factory):
-    root = tmp_path_factory.mktemp("corpus")
-    return SimpleNamespace(root=root, files=build_corpus(root))
-
-
-@pytest.fixture
-def env(tmp_path, corpus):
-    """A library holding the whole corpus, scanned into a fresh database."""
-    data = tmp_path / "data"
-    data.mkdir()
-    lib = tmp_path / "lib"
-    shutil.copytree(corpus.root, lib)
-    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)
-    scan = InventoryService(sf).scan(lib)
-    by_id = {
-        case.id: next(
-            (aid for path, aid in scan.asset_ids.items() if path.endswith(case.filename)), None
-        )
-        for case in CASES
-    }
-    yield SimpleNamespace(
-        config=config, lib=lib, sf=sf, scan=scan, assets=by_id,
-        thumbs=ThumbnailService(sf, config),
-    )
-    engine.dispose()
-
-
-# ── the manifest itself ───────────────────────────────────────────────────────
-
-
-def test_the_manifest_is_internally_consistent():
-    assert len({case.id for case in CASES}) == len(CASES), "logical ids must be unique"
-    assert len({case.filename for case in CASES}) == len(CASES), "paths must be unique"
-    for case in CASES:
-        assert case.kind in {"format", "orientation", "profile", "damaged", "metadata"}
-        assert case.expect_error in (None, "unsupported_image", "image_too_large")
-
-
-def test_the_corpus_regenerates_byte_for_byte(tmp_path):
-    """A golden corpus that drifts between runs cannot be a golden corpus."""
-    first = build_corpus(tmp_path / "one")
-    second = build_corpus(tmp_path / "two")
-    digests = {
-        case_id: (
-            hashlib.sha256(first[case_id].read_bytes()).hexdigest(),
-            hashlib.sha256(second[case_id].read_bytes()).hexdigest(),
-        )
-        for case_id in first
-    }
-    drifted = [case_id for case_id, (a, b) in digests.items() if a != b]
-    assert not drifted, f"non-deterministic fixtures: {drifted}"
-
-
-# ── bounded decode ────────────────────────────────────────────────────────────
-
-
-@pytest.mark.parametrize("case", RENDERABLE, ids=lambda case: case.id)
-def test_every_supported_case_renders(case, env):
-    asset_id = env.assets[case.id]
-    assert asset_id, f"{case.id} was not discovered by the scan"
-    path = env.thumbs.generate(asset_id, 256)
-    with Image.open(path) as thumb:
-        assert thumb.format == "WEBP"
-        assert max(thumb.size) <= 256
-        assert min(thumb.size) >= 1
-
-
-@pytest.mark.parametrize("orientation", range(1, 9))
-def test_exif_orientation_is_applied_before_resizing(orientation, env):
-    """All eight tags: the 90° ones must come out portrait from a landscape source."""
-    asset_id = env.assets[f"orientation_{orientation}"]
-    with Image.open(env.thumbs.generate(asset_id, 256)) as thumb:
-        rotated = orientation in (5, 6, 7, 8)
-        assert (thumb.height > thumb.width) is rotated
-
-
-def test_transparency_and_grayscale_survive_the_pipeline(env):
-    with Image.open(env.thumbs.generate(env.assets["png_alpha"], 256)) as thumb:
-        assert "A" in thumb.getbands()
-    with Image.open(env.thumbs.generate(env.assets["grayscale"], 256)) as thumb:
-        assert thumb.size == (100, 100)  # smaller than the request: never upscaled
-
-
-def test_a_broken_colour_profile_still_produces_a_picture(env):
-    """An unusable ICC profile is a metadata problem, not a reason to lose the preview."""
-    for case_id in ("icc_tagged", "broken_icc"):
-        with Image.open(env.thumbs.generate(env.assets[case_id], 256)) as thumb:
-            assert thumb.size[0] > 0
-
-
-def test_a_large_jpeg_is_not_decoded_at_full_resolution(env):
-    """12 megapixels would be ~36 MB of pixels; the draft decode keeps it far below."""
-    tracemalloc.start()
-    try:
-        env.thumbs.generate(env.assets["large_jpeg"], 256)
-        _, peak = tracemalloc.get_traced_memory()
-    finally:
-        tracemalloc.stop()
-    assert peak < 12_000_000, f"decode peaked at {peak} bytes"
-
-
-# ── damaged inputs ────────────────────────────────────────────────────────────
-
-
-@pytest.mark.parametrize("case", DAMAGED, ids=lambda case: case.id)
-def test_every_damaged_case_is_a_precise_item_error(case, env):
-    asset_id = env.assets[case.id]
-    assert asset_id, f"{case.id} was not discovered by the scan"
-    with pytest.raises(ThumbnailError) as raised:
-        env.thumbs.generate(asset_id, 256)
-    assert raised.value.code == case.expect_error
-    # Persisted, so a broken original is not re-decoded on every request...
-    with env.sf() as session:
-        rows = [
-            row
-            for row in session.query(Thumbnail).all()
-            if row.asset_id == asset_id and row.state == "error"
-        ]
-    assert rows and rows[0].error_code == case.expect_error
-    # ...and the cached failure is the same precise error, not a generic one.
-    with pytest.raises(ThumbnailError) as again:
-        env.thumbs.generate(asset_id, 256)
-    assert again.value.code == case.expect_error
-
-
-def test_a_gigapixel_header_is_refused_quickly_and_cheaply(env):
-    """The refusal must come from the declared size, not from decoding it."""
-    tracemalloc.start()
-    started = time.monotonic()
-    try:
-        with pytest.raises(ThumbnailError):
-            env.thumbs.generate(env.assets["bomb_header"], 1280)
-        _, peak = tracemalloc.get_traced_memory()
-    finally:
-        tracemalloc.stop()
-    assert time.monotonic() - started < 5
-    assert peak < 5_000_000, f"a 3.6 gigapixel header allocated {peak} bytes"
-
-
-def test_one_broken_file_stops_neither_the_scan_nor_its_neighbours(env):
-    """The whole corpus is in the library, damaged files included."""
-    assert len(env.scan.asset_ids) >= len(CASES) - 1  # the empty file has no pixels to hash
-    for case in RENDERABLE[:5]:
-        assert env.thumbs.generate(env.assets[case.id], 256).exists()
-
-
-def test_a_job_that_meets_a_broken_file_gets_evidence_not_an_exception(env):
-    """``ensure_protected`` is what the archive lane calls; a plan must not die on
-    one unreadable original."""
-    evidence = env.thumbs.ensure_protected(env.assets["corrupt_png"])
-    assert evidence["state"] == "unsupported"
-    assert evidence["error_code"] == "unsupported_image"
-    assert env.thumbs.ensure_protected(env.assets["jpeg"])["state"] == "ready"
-
-
-def test_undecodable_files_do_not_break_hashing(env):
-    """Discovery records what it can: byte identity always, pixel identity when the
-    file has pixels."""
-    with env.sf() as session:
-        assets = {a.current_path: a for a in session.query(Asset).all()}
-    for case in DAMAGED:
-        asset = next((a for path, a in assets.items() if path.endswith(case.filename)), None)
-        if asset is None:
-            continue
-        assert asset.current_sha256, "byte identity is always available"
-        assert asset.pixel_sha256 is None, "undecodable files must not invent pixel identity"
-
-
-# ── cache lifecycle ───────────────────────────────────────────────────────────
-
-
-def test_a_failed_render_leaves_no_temporary_behind(env):
-    with pytest.raises(ThumbnailError):
-        env.thumbs.generate(env.assets["truncated_jpeg"], 256)
-    leftovers = list(env.config.thumbnail_cache_dir.rglob(f"*{TEMP_SUFFIX}"))
-    assert leftovers == []
-
-
-def test_startup_cleanup_removes_only_recognised_temporaries(env):
-    cache = env.config.thumbnail_cache_dir
-    env.thumbs.generate(env.assets["jpeg"], 256)  # creates the cache directory
-    stale = cache / f".abc123{TEMP_SUFFIX}"
-    stale.write_bytes(b"half a thumbnail")
-    innocent = cache / "keep-me.webp"
-    innocent.write_bytes(b"not a temporary")
-
-    assert env.thumbs.cleanup_temp_files() == 1
-    assert not stale.exists()
-    assert innocent.exists()
-    assert list(cache.rglob("*.webp")), "real cache entries survive"
-
-
-def test_metadata_only_change_reuses_the_thumbnail_and_a_pixel_change_does_not(env):
-    """The cache key follows the pixels: an EXIF edit must not cost a re-render."""
-    asset_id = env.assets["jpeg"]
-    first = env.thumbs.generate(asset_id, 256)
-
-    with env.sf() as session:  # what a rescan records after an EXIF-only edit
-        asset = session.get(Asset, asset_id)
-        asset.current_sha256 = "different-bytes"
-        session.commit()
-    assert env.thumbs.generate(asset_id, 256) == first
-
-    with env.sf() as session:  # a genuine pixel change
-        asset = session.get(Asset, asset_id)
-        asset.pixel_sha256 = "different-pixels"
-        session.commit()
-    assert env.thumbs.generate(asset_id, 256) != first
-
-
-def test_a_deleted_cache_file_is_regenerated(env):
-    asset_id = env.assets["png"]
-    path = env.thumbs.generate(asset_id, 256)
-    path.unlink()
-    regenerated = env.thumbs.generate(asset_id, 256)
-    assert regenerated == path and regenerated.exists()
-
-
-# ── the imaging door itself ───────────────────────────────────────────────────
-
-
-def test_open_image_translates_every_decoder_failure(corpus):
-    """Callers must be able to catch two typed errors, never bare ``Exception``."""
-    for case in DAMAGED:
-        expected = imaging.ImageTooLarge if case.expect_error == "image_too_large" else imaging.UndecodableImage
-        with pytest.raises(expected):
-            with imaging.open_image(corpus.files[case.id]) as image:
-                image.load()
-
-
-def test_open_image_refuses_more_pixels_than_the_caller_allowed(corpus):
-    with pytest.raises(imaging.ImageTooLarge):
-        with imaging.open_image(corpus.files["jpeg"], max_pixels=100):
-            pass
-    with imaging.open_image(corpus.files["jpeg"], max_pixels=100_000) as image:
-        assert image.size == (320, 240)
-
-
-@needs_exiftool
-def test_malformed_metadata_does_not_stop_the_picture(env):
-    """A broken EXIF block is a metadata fact, not a decode failure."""
-    assert env.thumbs.generate(env.assets["malformed_metadata"], 256).exists()
-    with env.sf() as session:
-        asset = session.get(Asset, env.assets["malformed_metadata"])
-    assert asset.pixel_sha256, "pixels are still identifiable"
-
-
-def test_scoring_never_relaxes_truncated_image_handling(corpus):
-    """The donor's process-global ``LOAD_TRUNCATED_IMAGES`` is gone for good.
-
-    It is global state: switching it on for the safety model would also switch it on
-    for hashing and preview rendering in the same process, and half a file would
-    silently become a valid picture (donor_ledger: nt-score-model).
-    """
-    import inspect
-
-    from PIL import ImageFile
-
-    from photo_pipeline.integrations import nsfw_model
-
-    code = [
-        line
-        for line in inspect.getsource(nsfw_model).splitlines()
-        if not line.strip().startswith("#")
-    ]
-    assert not any("LOAD_TRUNCATED_IMAGES" in line for line in code)
-
-    # And the door holds even when something else in the process turned it on —
-    # the frozen donor does exactly that when the characterization suite imports it.
-    previous = ImageFile.LOAD_TRUNCATED_IMAGES
-    ImageFile.LOAD_TRUNCATED_IMAGES = True
-    try:
-        with pytest.raises(imaging.UndecodableImage):
-            with imaging.open_image(corpus.files["truncated_jpeg"]) as image:
-                image.load()
-        assert ImageFile.LOAD_TRUNCATED_IMAGES is True, "the caller's setting is restored"
-    finally:
-        ImageFile.LOAD_TRUNCATED_IMAGES = previous
-
-
-def test_every_manifest_case_is_exercised():
-    """The corpus lint: no fixture may sit in the manifest untested."""
-    covered = {case.id for case in RENDERABLE} | {case.id for case in DAMAGED}
-    assert covered == set(CASES_BY_ID)

tests/story_traceability.json

@@ -152,16 +152,6 @@
       "tests/unit/test_security_policy.py",
       "tests/integration/test_security_boundaries.py",
       "tests/e2e/test_security.py"
-    ],
-    "US07-03": [
-      "tests/integration/test_media_hardening.py",
-      "tests/integration/test_exif_checkpoints.py"
-    ],
-    "US07-04": [
-      "tests/unit/test_fault_artifacts.py",
-      "tests/integration/test_concurrency_races.py",
-      "tests/integration/test_fault_matrix.py",
-      "tests/e2e/test_crash_recovery.py"
     ]
   }
 }

tests/unit/test_fault_artifacts.py

@@ -1,69 +0,0 @@
-"""The failure-artifact collector (US07-04).
-
-CI must be able to explain a randomized failure after the temporary library is
-gone, which means the seed, the database, the journals, the logs, and a complete
-filesystem manifest have to leave the temporary directory before pytest cleans it.
-"""
-
-from __future__ import annotations
-
-import hashlib
-import json
-
-from tests._artifacts import collect, manifest
-
-
-def _library(root):
-    (root / "data").mkdir(parents=True)
-    (root / "data" / "photo_pipeline.db").write_bytes(b"database bytes")
-    (root / "data" / "photo_pipeline.db-wal").write_bytes(b"write ahead log")
-    (root / "data" / "uploads").mkdir()
-    (root / "data" / "uploads" / "batch.log").write_text("INFO uploaded a.jpg\n")
-    (root / "lib").mkdir()
-    (root / "lib" / "a.jpg").write_bytes(b"pixels")
-    return root
-
-
-def test_the_manifest_covers_every_file_with_its_hash(tmp_path):
-    root = _library(tmp_path / "run")
-    entries = {entry["path"]: entry for entry in manifest(root)}
-
-    assert set(entries) == {
-        "data/photo_pipeline.db",
-        "data/photo_pipeline.db-wal",
-        "data/uploads/batch.log",
-        "lib/a.jpg",
-    }
-    assert entries["lib/a.jpg"]["sha256"] == hashlib.sha256(b"pixels").hexdigest()
-    assert entries["lib/a.jpg"]["bytes"] == 6
-
-
-def test_collect_keeps_the_database_journals_logs_and_seed(tmp_path, monkeypatch):
-    root = _library(tmp_path / "run")
-    monkeypatch.setenv("PHOTO_PIPELINE_TEST_ARTIFACTS", str(tmp_path / "artifacts"))
-
-    destination = collect(root, "tests/x.py::test_races[run1]", properties={"race_seed": 1234})
-
-    seeds = json.loads((destination / "seeds.json").read_text())
-    assert seeds["properties"]["race_seed"] == 1234
-    assert seeds["test"].endswith("test_races[run1]")
-
-    kept = {str(p.relative_to(destination / "files")) for p in (destination / "files").rglob("*") if p.is_file()}
-    assert kept == {
-        "data/photo_pipeline.db",
-        "data/photo_pipeline.db-wal",
-        "data/uploads/batch.log",
-    }, "databases, write-ahead logs, and logs are the diagnosable evidence"
-    # The photo itself is never copied out of the library — but it is in the
-    # manifest, so a file that went missing is still provable.
-    assert any(entry["path"] == "lib/a.jpg" for entry in json.loads((destination / "manifest.json").read_text()))
-
-
-def test_collecting_twice_for_one_test_is_safe(tmp_path, monkeypatch):
-    root = _library(tmp_path / "run")
-    monkeypatch.setenv("PHOTO_PIPELINE_TEST_ARTIFACTS", str(tmp_path / "artifacts"))
-
-    first = collect(root, "tests/x.py::test_a")
-    second = collect(root, "tests/x.py::test_a")
-
-    assert first == second and (second / "manifest.json").exists()

tests/unit/test_rename_journal_states.py

@@ -77,8 +77,6 @@ def test_unsafe_states_are_the_ones_where_disk_may_have_changed():
         JournalState.MOVING,
         JournalState.MOVED,
         JournalState.DATABASE_UPDATED,
-        # The move happened and a human still has to decide about it (US07-04).
-        JournalState.ROLLBACK_REQUIRED,
     }
     # planned has not touched anything; complete/rolled_back are settled.
     assert JournalState.PLANNED not in UNSAFE_STATES