.gitignore

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

README.md

@@ -252,6 +252,35 @@ twice to prove it does not drift.
 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

photo_pipeline/faults.py

@@ -0,0 +1,45 @@
+"""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/integrations/exiftool.py

@@ -17,6 +17,20 @@ 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.
@@ -34,8 +48,9 @@ def read_keyword_sets(paths: Iterable[str]) -> dict[str, set[str]]:
             input="\n".join(paths),
             capture_output=True,
             text=True,
+            timeout=_timeout(),
         )
-    except FileNotFoundError:
+    except (FileNotFoundError, subprocess.TimeoutExpired):
         return {}
     out: dict[str, set[str]] = {}
     try:
@@ -67,9 +82,12 @@ def read_all(path: str) -> dict | None:
     """
     try:
         result = subprocess.run(
-            ["exiftool", "-m", "-j", "-G0:1", path], capture_output=True, text=True
+            ["exiftool", "-m", "-j", "-G0:1", path],
+            capture_output=True,
+            text=True,
+            timeout=_timeout(),
         )
-    except FileNotFoundError:
+    except (FileNotFoundError, subprocess.TimeoutExpired):
         return None
     try:
         records = json.loads(result.stdout or "[]")
@@ -91,4 +109,10 @@ def apply_keywords(path: str, *, add: Iterable[str] = (), remove: Iterable[str]
     if len(args) == 3:
         return True
     args.append(path)
-    return subprocess.run(args, capture_output=True, text=True).returncode == 0
+    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

photo_pipeline/jobs/worker.py

@@ -16,6 +16,7 @@ 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
@@ -86,6 +87,9 @@ 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/services/analysis.py

@@ -78,6 +78,11 @@ 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:
@@ -161,6 +166,22 @@ 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",

photo_pipeline/services/archive_transfer.py

@@ -52,6 +52,7 @@ 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,
@@ -63,7 +64,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, maybe_fault
+from photo_pipeline.services.rename_apply import PreconditionFailed
 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

@@ -25,6 +25,7 @@ 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
@@ -120,6 +121,10 @@ def run(
     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")

photo_pipeline/services/jobs.py

@@ -236,17 +236,32 @@ 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}")
-            job.state = to_state
+
-            job.version += 1
+            # Compare-and-set on the version this decision was made against. Without
-            job.updated_at = now
+            # 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,
+            }
             if error:
-                job.error_code, job.error_message = error
+                values["error_code"], values["error_message"] = error
             if to_state in TERMINAL_STATES:
-                job.finished_at = now
+                values.update(finished_at=now, lease_owner=None, lease_expires_at=None)
-                job.lease_owner = None
+            result = session.execute(
-                job.lease_expires_at = None
+                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"
+                )
             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,9 +46,11 @@ 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,
@@ -83,18 +85,6 @@ 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
@@ -143,20 +133,10 @@ class RenameApplyService:
                 self._apply_one(operation, token=token, worker_id=worker_id)
                 applied += 1
             except PreconditionFailed as error:
-                self.journal.transition(
+                self._record_failure(operation["id"], token, error.code, str(error))
-                    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.journal.transition(
+                self._record_failure(operation["id"], token, "apply_error", str(error))
-                    operation["id"],
-                    JournalState.FAILED,
-                    fencing_token=token,
-                    error=("apply_error", str(error)),
-                )
                 failed += 1
         state = self.journal.sync_plan_state(plan_id)
         return {
@@ -167,6 +147,26 @@ 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,7 +81,17 @@ ALLOWED_TRANSITIONS = {
 
 TERMINAL_STATES = frozenset({JournalState.COMPLETE, JournalState.ROLLED_BACK})
 # States where the disk may already have been touched by this operation.
-UNSAFE_STATES = frozenset({JournalState.MOVING, JournalState.MOVED, JournalState.DATABASE_UPDATED})
+# ``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,
+    }
+)
 
 RESUMABLE = "resumable"
 ROLLBACK_SAFE = "rollback_safe"

photo_pipeline/services/restores.py

@@ -53,6 +53,7 @@ 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
@@ -73,7 +74,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, maybe_fault
+from photo_pipeline.services.rename_apply import PreconditionFailed
 from photo_pipeline.services.rename_journal import RenameJournal
 
 PREFLIGHT_VERSION = 1

photo_pipeline/services/upload_batches.py

@@ -37,6 +37,7 @@ 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
@@ -236,6 +237,10 @@ 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

@@ -0,0 +1,96 @@
+"""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,5 +1,6 @@
-"""Make the repository root importable for the pipeline test suites, and give every
+"""Make the repository root importable for the pipeline test suites, give every
-suite the application session the API requires since US07-02.
+suite the application session the API requires since US07-02, and keep the evidence
+of any failure (US07-04).
 
 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,
@@ -99,3 +100,30 @@ 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

@@ -0,0 +1,274 @@
+"""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/integration/test_concurrency_races.py

@@ -0,0 +1,535 @@
+"""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_fault_matrix.py

@@ -0,0 +1,527 @@
+"""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/story_traceability.json

@@ -156,6 +156,12 @@
     "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

@@ -0,0 +1,69 @@
+"""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,6 +77,8 @@ 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