US07-05: Deliver Backup and Operational Recovery

.gitignore

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

README.md

@@ -228,6 +228,133 @@ 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.
+
+## Backup and recovery (US07-05)
+
+Backups go through SQLite's online backup API, never a file copy: with WAL enabled
+the `.db` file alone is missing every committed page still in the write-ahead log.
+Each backup is a directory under `data/backups/` holding the snapshot and a
+`manifest.json` describing it — schema revision, SHA-256, row counts, the archive
+media the library depends on, and which configuration was set. Secrets are recorded
+as `configured`, never as values, so a manifest is safe to attach to a bug report.
+
+```bash
+work_item/scripts/python -m photo_pipeline backup --reason before-upgrade --keep 7
+work_item/scripts/python -m photo_pipeline verify-backup data/backups/<name>
+work_item/scripts/python -m photo_pipeline diagnostics
+```
+
+The same is available at `GET /api/v1/diagnostics`, `GET|POST /api/v1/backups`,
+`GET /api/v1/backups/{name}/verify`, and `POST /api/v1/backups/prune`. **Restore is
+not an endpoint** — it replaces the state of an installation, so it belongs to a
+stopped one and a person at a terminal.
+
+### Integrity check
+
+`verify-backup` runs `PRAGMA integrity_check` (structure) *and*
+`PRAGMA foreign_key_check` (references), compares the snapshot's SHA-256 with the
+manifest, and re-counts every table the manifest recorded. Any mismatch — bit rot, a
+truncated copy, a "repaired" snapshot — fails the check, and `restore` refuses a
+backup that does not verify.
+
+### Restore drill
+
+1. Stop the server and the worker.
+2. `python -m photo_pipeline verify-backup data/backups/<name>` — never restore an
+   unverified snapshot.
+3. `python -m photo_pipeline restore data/backups/<name> --into /path/to/fresh-data`
+   (a target that already holds a database is refused; recovering in place means
+   moving the old data directory aside first).
+4. Point `PHOTO_PIPELINE_DATA_DIR` at the restored directory and run
+   `python -m photo_pipeline migrate`.
+5. Run an inventory scan so paths are reconciled against the real library.
+6. Mount every archive location named in the manifest before archiving again — the
+   database records where archived originals are, but it does not contain them.
+
+Practise this against a copy before you need it; the drill is exercised
+automatically by `tests/integration/test_backup_recovery.py`.
+
+### Failed migration
+
+A pending schema upgrade is snapshotted first (`reason: pre-migration`), by both the
+API startup and `python -m photo_pipeline migrate`. If a migration fails, the error
+log names the backup directory: stop everything and run the restore drill against
+it. An up-to-date database is not backed up again on every start.
+
+### Archive media
+
+Archived originals live on their medium, not in the backup. The manifest lists every
+archive location with its `media_id` and whether it was mounted when the backup was
+taken. Keep one copy of each medium off-site, and remount a location before
+restoring assets from it.
+
+### Retention and disk
+
+`--keep N` (default 7) prunes the oldest backups and never the newest.
+`diagnostics` reports the database, write-ahead log, thumbnail cache, uploader
+reports, backups, and logs separately, with free space and warnings for low disk
+(`disk_low`, `disk_critical`), a cache over its quota, a write-ahead log outgrowing
+its database, and a legacy CLI writing the library.
+
+### Process locking
+
+`serve` and `worker` take a JSON lock in the data directory (`api.lock.json`,
+`worker.lock.json`). A second worker exits `2` and names the holder; a lock whose
+process is gone is taken over. If the frozen CLI's state files are being written,
+both refuse with exit `3` — `--allow-legacy` overrides, and you own the outcome.
+
 ## Legacy CLI archive
 
 The command-line tools this application was extracted from are frozen in

delivery_backlog/E08-container-deployment.md

@@ -0,0 +1,32 @@
+# E08 — Container Deployment
+
+Concept phase: none. This epic is a delivery-format addition on top of the concept:
+the same application, same safety invariants, packaged as a Docker image and deployed
+continuously from Gitea Actions instead of being started by hand from a working copy.
+
+It does not change the product scope in
+[`INTEGRATED_PIPELINE_CONCEPT.md`](../INTEGRATED_PIPELINE_CONCEPT.md). SQLite stays the
+store, one worker stays the writer, the library process lock stays authoritative, and
+no path outside the configured library roots becomes reachable because the process now
+runs in a container.
+
+One decision does extend the concept and is made here explicitly: the application may
+be reached through a reverse proxy under a real hostname, not only over loopback. That
+requires a configurable trust boundary and an authentication gate, because the
+loopback-only checks of US07-02 are what currently stand in for authentication.
+
+## Stories
+
+1. [US08-01 — Make the trust boundary configurable and authenticated](stories/US08-01-trusted-hosts-auth.md)
+2. [US08-02 — Build a reproducible application image](stories/US08-02-container-image.md)
+3. [US08-03 — Compose the runtime and mount the library safely](stories/US08-03-compose-runtime.md)
+4. [US08-04 — Publish and deploy from Gitea Actions](stories/US08-04-gitea-cicd.md)
+5. [US08-05 — Automate container deployment acceptance](stories/US08-05-container-e2e.md)
+
+## Epic outcome
+
+A tagged image built from `main` runs the API and the worker as separate containers
+against a mounted library and a persistent data volume, is published to the Gitea
+registry, is redeployed by webhook, survives restart and upgrade with its database and
+journals intact, and refuses every request that a loopback deployment would have
+refused.

delivery_backlog/README.md

@@ -2,11 +2,13 @@
 
 This backlog decomposes the phases in
 [`INTEGRATED_PIPELINE_CONCEPT.md`](../INTEGRATED_PIPELINE_CONCEPT.md) into seven
-epics and small, independently verifiable user stories.
+epics and small, independently verifiable user stories, plus one delivery-format
+epic (E08) that packages the released application as a deployable container.
 
 ## Numbering and file naming
 
-- Epics: `E01` through `E07`, matching concept Phases A through G.
+- Epics: `E01` through `E07`, matching concept Phases A through G; `E08` has no
+  concept phase and must not change product scope.
 - Stories: `US<epic>-<sequence>`, for example `US03-02`.
 - Epic files: `E01-<slug>.md`.
 - Story files: `stories/US01-01-<slug>.md`.
@@ -36,6 +38,7 @@ epics and small, independently verifiable user stories.
 5. [E05 — Immich upload](E05-immich-upload.md)
 6. [E06 — Archive lifecycle](E06-archive-lifecycle.md)
 7. [E07 — Hardening and release](E07-hardening-release.md)
+8. [E08 — Container deployment](E08-container-deployment.md)
 
 ## Shared definition of done
 

delivery_backlog/stories/US08-01-trusted-hosts-auth.md

@@ -0,0 +1,42 @@
+# US08-01 — Make the Trust Boundary Configurable and Authenticated
+
+Epic: [E08](../E08-container-deployment.md)
+
+As an operator, I want to reach the application through my own hostname without
+weakening it, so a container behind a reverse proxy is as safe as the loopback
+deployment it replaces.
+
+## Context
+
+`photo_pipeline/api/security.py` refuses any request whose `Host` or `Origin` is not
+loopback. That check is the current stand-in for authentication: whoever can reach
+`127.0.0.1:8000` is the owner. Behind a proxy the hostname is no longer loopback, so
+relaxing the check without adding an authentication gate would publish the library.
+
+## Acceptance criteria
+
+- Allowed hosts and origins come from configuration (`PHOTO_PIPELINE_*`), default to
+  the current loopback set, and an unset configuration behaves exactly as today.
+- Whenever a non-loopback host is configured, startup requires an access secret and
+  refuses to serve without one; loopback-only deployments keep working with no secret.
+- The secret is exchanged for the existing session cookie and CSRF token through the
+  bootstrap endpoint; every protected route keeps its current session and CSRF
+  requirements unchanged.
+- Forwarded headers (`X-Forwarded-Proto`, `X-Forwarded-Host`) are honored only from a
+  configured trusted proxy and ignored otherwise, so a client cannot forge its origin.
+- Cookies are marked `Secure` when the effective external scheme is HTTPS.
+- Failed authentication is rate-limited and logged without the secret, the session id,
+  or any request body.
+- Health endpoints stay reachable without the secret; nothing else does.
+
+## Automated tests
+
+- Unit tests for host/origin evaluation across loopback default, configured host,
+  unconfigured host, forged forwarded headers, and trusted-proxy forwarded headers.
+- Integration tests: startup refusal without a secret, successful exchange, wrong
+  secret, replay of an old session, cross-site request, and unauthenticated access to
+  every route class.
+
+## Dependencies
+
+- US07-02

delivery_backlog/stories/US08-02-container-image.md

@@ -0,0 +1,41 @@
+# US08-02 — Build a Reproducible Application Image
+
+Epic: [E08](../E08-container-deployment.md)
+
+As an operator, I want one image that can run either application role, so deployment is
+a pull instead of a Python environment I have to reproduce by hand.
+
+## Context
+
+The application shells out to `exiftool` and `immich-go`, writes into the library as a
+normal filesystem user, and serves a static frontend from `frontend/`. All three have to
+be true inside the image, or the container starts and then fails on the first real
+operation.
+
+## Acceptance criteria
+
+- A `Dockerfile` builds from a pinned Python base, installs the project and its runtime
+  dependencies, and contains no test, playwright, or build-only tooling in the final
+  layer.
+- `exiftool` and `immich-go` are present at pinned versions, and their versions are
+  recorded in the image and reported by `python -m photo_pipeline diagnostics`.
+- The image runs as a non-root user whose UID/GID are build-time arguments, so files
+  the application renames or writes keep the ownership the host library expects.
+- One entrypoint selects the role: `serve` or `worker`, passing through the existing
+  CLI arguments; no supervisor runs two roles in one container.
+- `serve` containers declare a `HEALTHCHECK` against `/api/v1/health/ready`, so an
+  unmigrated or misconfigured database is not reported healthy.
+- The image contains no secrets, no library data, no database, and no `.git`; the build
+  context is constrained by `.dockerignore`.
+- Image build is reproducible from a clean checkout and documented in `README.md`.
+
+## Automated tests
+
+- A build-and-run test asserts the image starts, reports ready, serves the frontend
+  index, and returns the pinned `exiftool` and `immich-go` versions.
+- A test asserts the container refuses to run as UID 0 and that a file created by the
+  container is owned by the configured UID/GID.
+
+## Dependencies
+
+- US07-05

delivery_backlog/stories/US08-03-compose-runtime.md

@@ -0,0 +1,47 @@
+# US08-03 — Compose the Runtime and Mount the Library Safely
+
+Epic: [E08](../E08-container-deployment.md)
+
+As an operator, I want a single compose file that runs the API and the worker against my
+real library, so a deployment is one command and the safety invariants survive it.
+
+## Context
+
+The library process lock (US07-05) assumes both roles see the same lock file, and SQLite
+in WAL mode assumes a real local filesystem. Container path policy is the same problem
+as host path policy with a new failure mode: the configured library roots must name the
+in-container mount paths, not the host paths.
+
+## Acceptance criteria
+
+- `docker-compose.yml` runs exactly one `serve` and one `worker` container from the same
+  image and the same data volume, and a second worker is refused by the existing lock
+  rather than by convention.
+- The library is a bind mount; `PHOTO_PIPELINE_LIBRARY_ROOTS` names the container-side
+  paths, and a mismatch between mounted and configured roots fails at startup with a
+  clear message instead of at the first write.
+- The data volume holds the database, WAL, thumbnail cache, and backups on a local
+  filesystem; the composition documents that a network mount is unsupported for it.
+- Migrations run before `serve` and `worker` accept work, using the existing backup-then-
+  migrate path, and an upgrade that fails leaves the previous database intact.
+- Configuration and secrets come from the environment, never from the image or a
+  committed file; a `.env.example` lists every `PHOTO_PIPELINE_*` variable with safe
+  defaults and no values.
+- The API port is published to host loopback by default; exposing it publicly requires
+  the configured hostname and access secret from US08-01.
+- Containers restart automatically, and a restart mid-job resumes exactly as a host
+  restart does today.
+- Backup, verify-backup, restore, and diagnostics are documented as container commands
+  and work against the mounted volumes.
+
+## Automated tests
+
+- An integration test brings the composition up against a temporary fixture library,
+  runs a job, restarts both containers, and asserts the job resumes and the database is
+  intact.
+- Tests for: second worker refused, library-root mismatch refused at startup, failed
+  migration leaving the previous database restorable.
+
+## Dependencies
+
+- US08-01, US08-02

delivery_backlog/stories/US08-04-gitea-cicd.md

@@ -0,0 +1,40 @@
+# US08-04 — Publish and Deploy from Gitea Actions
+
+Epic: [E08](../E08-container-deployment.md)
+
+As a release owner, I want `main` to build, publish, and redeploy the image
+automatically, so deployment is the same reproducible path every time.
+
+## Context
+
+The workflow is adapted from the `crowdsec-admin` deployment workflow
+(`.gitea/workflows/deploy.yml` in that repository): build, log in to the Gitea registry,
+push, trigger a Portainer webhook, prune. This project needs the same shape plus a test
+gate, because unlike that project it has a required suite that must not be skipped.
+
+## Acceptance criteria
+
+- `.gitea/workflows/` contains a test workflow that runs on pull requests and on `main`,
+  executing the configured required suites, and a deploy workflow that runs only after
+  the tests pass on `main` and on manual dispatch.
+- The deploy workflow publishes to `git.domverse-berlin.eu` under this project's own
+  image path, tagged `latest` and the commit SHA, so a rollback is a tag change.
+- Registry credentials and the Portainer webhook come from repository secrets; runtime
+  secrets (vision key, Immich key, access secret) stay in the Portainer stack and never
+  enter the repository or the image.
+- Redeploy is triggered by webhook and the workflow fails when the webhook call fails.
+- Dangling images are pruned; published tags are not.
+- A concurrency guard prevents two deploys of different commits overlapping.
+- `README.md` documents the required secrets, the image path, the rollback procedure,
+  and that the stack is managed by Portainer from git.
+
+## Automated tests
+
+- Workflow files are validated (syntax and required job/step names) by a repository test
+  so a rename cannot silently disable the test gate.
+- A dry-run job builds and pushes to a scratch tag on manual dispatch without touching
+  `latest` or triggering a redeploy.
+
+## Dependencies
+
+- US08-02, US08-03

delivery_backlog/stories/US08-05-container-e2e.md

@@ -0,0 +1,30 @@
+# US08-05 — Automate Container Deployment Acceptance
+
+Epic: [E08](../E08-container-deployment.md)
+
+As a release owner, I want one automated gate that proves the deployed container, so the
+packaged application is verified the same way the host application is.
+
+## Acceptance criteria
+
+- One documented command provisions the composition from the built image against a
+  temporary fixture library and an isolated data volume, and destroys it afterwards.
+- A browser journey against the containerized application covers discovery, duplicate
+  review, analysis, album proposal, rename, upload preflight, and archive views.
+- An upgrade journey runs the previous published image, then the new one, and asserts
+  migrations, journals, jobs, and the thumbnail cache survive.
+- A restart journey kills both containers mid-job and asserts resume without duplicate
+  side effects.
+- Security gates run against the deployed instance: unauthenticated access refused,
+  forged forwarded headers refused, paths outside the mounted library roots refused, and
+  no secret in container logs.
+- Evidence is retained per run and the gate fails on any skipped required check.
+
+## Automated tests
+
+- The container acceptance suite runs on a `phase_h` marker in CI on `main` and before a
+  published deploy; earlier epic suites keep running unchanged.
+
+## Dependencies
+
+- US08-01 through US08-04

legacy_cli_archive/donor_ledger.yaml

@@ -480,6 +480,13 @@ 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

@@ -0,0 +1,42 @@
+"""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/__main__.py

@@ -1,21 +1,36 @@
-"""Application management CLI: ``python -m photo_pipeline {serve,migrate,worker,import-legacy-scores}``."""
+"""Application management CLI:
+``python -m photo_pipeline {serve,migrate,worker,import-legacy-scores,backup,verify-backup,restore,diagnostics}``.
+
+``serve`` and ``worker`` take the library process lock for their role (US07-05):
+two workers, or the frozen CLI running beside the app, would each be safe on their
+own and destructive together. ``restore`` is here rather than in the API because it
+replaces the state of an installation and belongs to a stopped one.
+"""
 
 from __future__ import annotations
 
 import argparse
+import json
 from typing import Sequence
 
 from photo_pipeline.config import Config
-from photo_pipeline.db import run_migrations
+from photo_pipeline.services.app_lock import LegacyProcessActive, LibraryLock, LockHeld
+from photo_pipeline.services.backup import BackupError, BackupService, migrate_with_backup
 
 
 def main(argv: Sequence[str] | None = None) -> int:
     parser = argparse.ArgumentParser(prog="photo_pipeline")
     commands = parser.add_subparsers(dest="command", required=True)
-    commands.add_parser("serve", help="Run the API server")
+    serve_cmd = commands.add_parser("serve", help="Run the API server")
     commands.add_parser("migrate", help="Upgrade the database to the latest revision")
     worker_cmd = commands.add_parser("worker", help="Run a durable-job worker")
     worker_cmd.add_argument("--id", default="worker-1", help="Worker id (lease owner)")
+    for locked in (serve_cmd, worker_cmd):
+        locked.add_argument(
+            "--allow-legacy",
+            action="store_true",
+            help="Start even though a legacy CLI looks active (unsafe; you own the outcome)",
+        )
     import_cmd = commands.add_parser(
         "import-legacy-scores",
         help="Import the archived CLI's nsfw_scores.csv into the database (US07-01)",
@@ -27,22 +42,64 @@ def main(argv: Sequence[str] | None = None) -> int:
     import_cmd.add_argument(
         "--dry-run", action="store_true", help="Report what would happen and change nothing"
     )
+
+    backup_cmd = commands.add_parser("backup", help="Take an online database backup")
+    backup_cmd.add_argument("--reason", default="manual", help="Why (part of the directory name)")
+    backup_cmd.add_argument("--keep", type=int, default=7, help="How many backups to retain")
+    verify_cmd = commands.add_parser("verify-backup", help="Check a backup is intact and readable")
+    verify_cmd.add_argument("backup", help="Path to the backup directory")
+    restore_cmd = commands.add_parser(
+        "restore", help="Restore a verified backup into a fresh data directory"
+    )
+    restore_cmd.add_argument("backup", help="Path to the backup directory")
+    restore_cmd.add_argument("--into", required=True, help="Fresh data directory to restore into")
+    commands.add_parser("diagnostics", help="Report sizes, disk headroom, locks, and warnings")
+
     args = parser.parse_args(argv)
 
     config = Config.from_env()
     config.database_path.parent.mkdir(parents=True, exist_ok=True)
 
     if args.command == "migrate":
-        run_migrations(config.database_url)
+        manifest = migrate_with_backup(config)
+        if manifest:
+            print(json.dumps({"pre_migration_backup": manifest["name"]}, indent=2))
+        return 0
+
+    if args.command == "backup":
+        try:
+            manifest = BackupService(config).create(reason=args.reason, keep=args.keep)
+        except BackupError as error:
+            print(str(error))
+            return 1
+        print(json.dumps(manifest, indent=2))
+        return 0
+
+    if args.command == "verify-backup":
+        result = BackupService(config).verify(args.backup)
+        print(json.dumps(result.as_dict(), indent=2))
+        return 0 if result.ok else 1
+
+    if args.command == "restore":
+        try:
+            report = BackupService(config).restore(args.backup, args.into)
+        except BackupError as error:
+            print(str(error))
+            return 1
+        print(json.dumps(report, indent=2))
+        return 0
+
+    if args.command == "diagnostics":
+        from photo_pipeline.services import diagnostics
+
+        print(json.dumps(diagnostics.report(config), indent=2))
         return 0
 
     if args.command == "import-legacy-scores":
-        import json
-
         from photo_pipeline.db import create_db_engine, create_session_factory
         from photo_pipeline.services.legacy_import import LegacyImportService, write_report
 
-        run_migrations(config.database_url)
+        migrate_with_backup(config)
         engine = create_db_engine(config.database_url)
         service = LegacyImportService(create_session_factory(engine))
         report = service.import_nsfw_scores(
@@ -63,18 +120,50 @@ def main(argv: Sequence[str] | None = None) -> int:
         import photo_pipeline.jobs.domain_handlers  # noqa: F401
         from photo_pipeline.jobs.worker import Worker
 
-        run_migrations(config.database_url)
-        engine = create_db_engine(config.database_url)
-        Worker(create_session_factory(engine), worker_id=args.id, config=config).run_forever()
+        lock = LibraryLock(config, "worker")
+        if (held := _acquire(lock, allow_legacy=args.allow_legacy)) is not None:
+            return held
+        try:
+            migrate_with_backup(config)
+            engine = create_db_engine(config.database_url)
+            Worker(
+                create_session_factory(engine), worker_id=args.id, config=config
+            ).run_forever()
+        finally:
+            lock.release()
         return 0
 
     import uvicorn
 
     from photo_pipeline.api.app import create_app
 
-    uvicorn.run(create_app(config), host=config.host, port=config.port)
+    lock = LibraryLock(config, "api")
+    if (held := _acquire(lock, allow_legacy=args.allow_legacy)) is not None:
+        return held
+    try:
+        uvicorn.run(create_app(config), host=config.host, port=config.port)
+    finally:
+        lock.release()
     return 0
 
 
+def _acquire(lock: LibraryLock, *, allow_legacy: bool) -> int | None:
+    """Take the lock, or explain on stderr why this process must not start.
+
+    Returns an exit code to return, or ``None`` when the lock was acquired.
+    """
+    import sys
+
+    try:
+        lock.acquire(allow_legacy=allow_legacy)
+    except LockHeld as error:
+        print(str(error), file=sys.stderr)
+        return 2
+    except LegacyProcessActive as error:
+        print(f"{error} (override with --allow-legacy)", file=sys.stderr)
+        return 3
+    return None
+
+
 if __name__ == "__main__":
     raise SystemExit(main())

photo_pipeline/api/app.py

@@ -27,6 +27,7 @@ from photo_pipeline.api.routes import (
     inventory,
     jobs,
     library,
+    operations,
     renames,
     safety,
     session as session_routes,
@@ -39,8 +40,10 @@ from photo_pipeline.api.security import DEFAULT_HEADERS, SecurityMiddleware, Ses
 # Registers the safety_score / analysis job handlers on import.
 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.db import create_db_engine, create_session_factory
+from photo_pipeline.services.backup import migrate_with_backup
 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"
@@ -88,7 +91,9 @@ def create_app(config: Config | None = None) -> FastAPI:
     @asynccontextmanager
     async def lifespan(app: FastAPI):
         config.database_path.parent.mkdir(parents=True, exist_ok=True)
-        run_migrations(config.database_url)
+        # A schema upgrade is snapshotted first, so a migration that fails halfway
+        # leaves a restorable database behind rather than a damaged one (US07-05).
+        migrate_with_backup(config)
         engine = create_db_engine(config.database_url)
         app.state.config = config
         app.state.engine = engine
@@ -96,6 +101,9 @@ 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:
@@ -122,6 +130,7 @@ def create_app(config: Config | None = None) -> FastAPI:
     app.include_router(renames.router, prefix="/api/v1")
     app.include_router(uploads.router, prefix="/api/v1")
     app.include_router(archives.router, prefix="/api/v1")
+    app.include_router(operations.router, prefix="/api/v1")
     # Static single-page app (hash-routed). Mounted last so /api/v1 wins.
     if FRONTEND_DIR.is_dir():
         app.mount("/app", StaticFiles(directory=FRONTEND_DIR, html=True), name="app")

photo_pipeline/api/routes/operations.py

@@ -0,0 +1,70 @@
+"""Operational endpoints: diagnostics and backups (US07-05).
+
+Backups can be taken and verified here because both are safe, additive, and the
+operator needs them from the same screen that shows the disk filling up.
+
+**Restore is deliberately not an endpoint.** It replaces the state of the running
+application with an older one, so it belongs to a stopped installation and a person
+at a terminal: ``python -m photo_pipeline restore``. An HTTP call that can silently
+roll the library back to last week is a hole, not a feature.
+"""
+
+from __future__ import annotations
+
+from fastapi import APIRouter, Query, Request
+from fastapi.responses import JSONResponse
+from pydantic import BaseModel
+
+from photo_pipeline.services import diagnostics
+from photo_pipeline.services.backup import DEFAULT_KEEP, BackupError, BackupService
+
+router = APIRouter(tags=["operations"])
+
+
+class CreateBackupRequest(BaseModel):
+    reason: str = "manual"
+    keep: int = DEFAULT_KEEP
+
+
+def _service(request: Request) -> BackupService:
+    return BackupService(request.app.state.config)
+
+
+def _error(status: int, code: str, message: str) -> JSONResponse:
+    return JSONResponse(status_code=status, content={"error": {"code": code, "message": message}})
+
+
+@router.get("/diagnostics")
+def read_diagnostics(request: Request) -> dict:
+    return diagnostics.report(request.app.state.config)
+
+
+@router.get("/backups")
+def list_backups(request: Request) -> dict:
+    return {"backups": _service(request).list()}
+
+
+@router.post("/backups", status_code=201)
+def create_backup(body: CreateBackupRequest, request: Request):
+    try:
+        return _service(request).create(reason=body.reason, keep=body.keep)
+    except BackupError as error:
+        return _error(422, "backup_failed", str(error))
+
+
+@router.get("/backups/{name}/verify")
+def verify_backup(name: str, request: Request):
+    service = _service(request)
+    # The name comes from the browser, so it names a backup — it is never joined
+    # into a path until it has been matched against one that exists (US07-02).
+    if name not in {entry["name"] for entry in service.list()}:
+        return _error(404, "not_found", f"unknown backup {name}")
+    return {"name": name, **service.verify(service.root / name).as_dict()}
+
+
+@router.post("/backups/prune")
+def prune_backups(request: Request, keep: int = Query(DEFAULT_KEEP, ge=1)):
+    try:
+        return {"removed": _service(request).prune(keep=keep)}
+    except BackupError as error:
+        return _error(422, "invalid_retention", str(error))

photo_pipeline/db.py

@@ -41,12 +41,41 @@ def create_session_factory(engine: Engine) -> sessionmaker:
     return sessionmaker(bind=engine, expire_on_commit=False, future=True)
 
 
-def run_migrations(url: str) -> None:
-    """Upgrade the database at ``url`` to the latest revision."""
-    from alembic import command
+def _alembic_config(url: str):
     from alembic.config import Config as AlembicConfig
 
     cfg = AlembicConfig(str(_REPO_ROOT / "alembic.ini"))
     cfg.set_main_option("script_location", str(_REPO_ROOT / "migrations"))
     cfg.set_main_option("sqlalchemy.url", url)
-    command.upgrade(cfg, "head")
+    return cfg
+
+
+def run_migrations(url: str) -> None:
+    """Upgrade the database at ``url`` to the latest revision."""
+    from alembic import command
+
+    command.upgrade(_alembic_config(url), "head")
+
+
+def head_revision() -> str | None:
+    """The revision this code expects. ``None`` if the scripts cannot be read."""
+    from alembic.script import ScriptDirectory
+
+    try:
+        return ScriptDirectory.from_config(_alembic_config("sqlite://")).get_current_head()
+    except Exception:
+        return None
+
+
+def current_revision(url: str) -> str | None:
+    """The revision a database is actually at, or ``None`` for an unstamped one."""
+    engine = create_db_engine(url)
+    try:
+        with engine.connect() as connection:
+            from alembic.runtime.migration import MigrationContext
+
+            return MigrationContext.configure(connection).get_current_revision()
+    except Exception:
+        return None
+    finally:
+        engine.dispose()

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/imaging.py

@@ -0,0 +1,119 @@
+"""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,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:
@@ -56,6 +71,33 @@ 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"]
@@ -67,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/integrations/nsfw_model.py

@@ -15,6 +15,8 @@ from __future__ import annotations
 
 from pathlib import Path
 
+from photo_pipeline import imaging
+
 MODEL_ID = "AdamCodd/vit-base-nsfw-detector"
 BATCH = 16
 
@@ -54,13 +56,17 @@ class NsfwModel:
         self._ensure_loaded()
         import numpy as np
         import torch
-        from PIL import Image, ImageFile
+        from PIL import Image
 
-        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
+        # 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
             return torch.from_numpy(array).permute(2, 0, 1)
 
         results: list[tuple[str, float]] = []
@@ -69,9 +75,10 @@ class NsfwModel:
             tensors, batch_paths = [], []
             for path in items[start : start + self.batch]:
                 try:
-                    tensors.append(preprocess(Image.open(path)))
+                    tensors.append(preprocess(path))
                     batch_paths.append(path)
-                except Exception:
+                except (imaging.MediaError, OSError, ValueError):
+                    # One bad file must not cost the batch its other fifteen.
                     continue
             if not tensors:
                 continue

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/models/__init__.py

@@ -7,6 +7,7 @@ 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,
@@ -28,6 +29,7 @@ __all__ = [
     "DuplicateCluster",
     "DuplicateMember",
     "DuplicateNegativeLink",
+    "ExifProjection",
     "Job",
     "JobItem",
     "JobEvent",

photo_pipeline/models/exif.py

@@ -0,0 +1,33 @@
+"""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,9 +26,8 @@ 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 hashing
+from photo_pipeline.services import exif_checkpoint
 from photo_pipeline.services.safety import SFW
 
 MODEL = "gemini-2.5-flash"
@@ -79,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:
@@ -162,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",
@@ -170,7 +190,7 @@ class AnalysisService:
                 tokens=int(result.get("_tokens", 0)) if isinstance(result, dict) else 0,
                 raw=json.dumps(result, ensure_ascii=False),
             )
-            _write_analysis_exif(path, result)
+            self._write_analysis_exif(asset_id, path, result)
             analyzed += 1
         return {"analyzed": analyzed, "skipped": skipped, "errors": errors}
 
@@ -195,10 +215,42 @@ 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)
@@ -211,17 +263,6 @@ 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/app_lock.py

@@ -0,0 +1,221 @@
+"""Library-level process lock, and detection of an incompatible legacy run
+(US07-05, concept §15 "migration and operational risks").
+
+Every safety this application has — durable job leases, rename journals, archive
+manifests — assumes that one installation owns the library. Two workers, or the
+frozen CLI running beside the app, break that assumption *below* the level those
+mechanisms can see: the second process simply does not know the first one's
+database exists.
+
+So mutation requires a file lock in the data directory, shaped as JSON so any
+future or migrated entry point can read and honour it without importing this
+package:
+
+    {"lock_version": 1, "role": "worker", "pid": 4242, "host": "...",
+     "started_at": "...", "library_roots": ["..."]}
+
+One holder per role: an API and a worker are designed to run together, a second
+worker is not. A lock whose process is gone is stale and is taken over with the
+takeover recorded — refusing to start because of a crashed predecessor would turn
+one outage into two.
+
+Legacy detection is deliberately a heuristic, not a promise: the archived CLI has
+no lock of its own, so what can be observed is its state files being written right
+now. Recent writes to them mean something else is mutating this library, and every
+mutating stage should refuse until it stops.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import socket
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from pathlib import Path
+
+from photo_pipeline.config import Config
+
+LOCK_VERSION = 1
+LOCK_SUFFIX = ".lock.json"
+# State files only the archived CLIs write. Their presence is history; a *recent*
+# modification is a running process.
+# ponytail: the real fix is a lock the migrated CLI paths take too — this catches
+# the frozen archive, which has no lock and cannot be changed (US07-01).
+LEGACY_ARTIFACTS = (
+    "photo_analyzer.db",
+    "nsfw_scores.csv",
+    "photo_analyzer_history.jsonl",
+    "photo_analyzer.log",
+    "photo_analyzer_debug.log",
+)
+LEGACY_ACTIVE_SECONDS = 300
+
+
+class LockHeld(RuntimeError):
+    """Another live process of the same role owns this library."""
+
+    def __init__(self, holder: "Holder") -> None:
+        super().__init__(
+            f"{holder.role} is already running for this library "
+            f"(pid {holder.pid} on {holder.host}, since {holder.started_at})"
+        )
+        self.holder = holder
+
+
+class LegacyProcessActive(RuntimeError):
+    """A legacy CLI appears to be mutating the same library right now."""
+
+
+@dataclass(frozen=True)
+class Holder:
+    role: str
+    pid: int
+    host: str
+    started_at: str
+    lock_version: int = LOCK_VERSION
+    library_roots: tuple[str, ...] = ()
+
+    @property
+    def alive(self) -> bool:
+        """Whether the recorded process still exists on this host.
+
+        A lock from another host cannot be probed, so it is believed: assuming a
+        remote holder is dead is how two machines end up renaming the same folder.
+        """
+        if self.host != socket.gethostname():
+            return True
+        try:
+            os.kill(self.pid, 0)
+        except ProcessLookupError:
+            return False
+        except PermissionError:
+            return True  # exists, owned by someone else
+        return True
+
+    def as_dict(self) -> dict:
+        return {
+            "lock_version": self.lock_version,
+            "role": self.role,
+            "pid": self.pid,
+            "host": self.host,
+            "started_at": self.started_at,
+            "library_roots": list(self.library_roots),
+            "alive": self.alive,
+        }
+
+
+def _now() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+def legacy_activity(config: Config) -> dict:
+    """Legacy state files written within the activity window, if any."""
+    seen: list[dict] = []
+    cutoff = _now().timestamp() - LEGACY_ACTIVE_SECONDS
+    roots = [Path(root) for root in config.library_roots] + [Path(config.data_dir)]
+    for root in roots:
+        for name in LEGACY_ARTIFACTS:
+            path = root / name
+            try:
+                modified = path.stat().st_mtime
+            except OSError:
+                continue
+            if modified >= cutoff:
+                seen.append(
+                    {
+                        "path": str(path),
+                        "modified_at": datetime.fromtimestamp(modified, timezone.utc).isoformat(),
+                    }
+                )
+    return {"active": bool(seen), "artifacts": seen, "window_seconds": LEGACY_ACTIVE_SECONDS}
+
+
+class LibraryLock:
+    """One holder per role for one library. Used as a context manager."""
+
+    def __init__(self, config: Config, role: str = "worker") -> None:
+        self._config = config
+        self.role = role
+        self.path = Path(config.data_dir) / f"{role}{LOCK_SUFFIX}"
+        self._acquired = False
+
+    # ── inspection ────────────────────────────────────────────────────────────
+
+    def holder(self) -> Holder | None:
+        try:
+            payload = json.loads(self.path.read_text())
+        except (OSError, ValueError):
+            return None
+        try:
+            return Holder(
+                role=payload["role"],
+                pid=int(payload["pid"]),
+                host=payload["host"],
+                started_at=payload["started_at"],
+                lock_version=int(payload.get("lock_version", LOCK_VERSION)),
+                library_roots=tuple(payload.get("library_roots", ())),
+            )
+        except (KeyError, TypeError, ValueError):
+            # An unreadable lock is not an absent lock: something wrote it.
+            return Holder(role=self.role, pid=-1, host="unknown", started_at="unknown")
+
+    # ── acquire / release ─────────────────────────────────────────────────────
+
+    def acquire(self, *, allow_legacy: bool = False) -> Holder:
+        """Take the lock for this role, or explain who has it.
+
+        Raises ``LockHeld`` when a live process of the same role owns the library,
+        and ``LegacyProcessActive`` when the archived CLI looks like it is running
+        against it.
+        """
+        if not allow_legacy:
+            legacy = legacy_activity(self._config)
+            if legacy["active"]:
+                raise LegacyProcessActive(
+                    "a legacy CLI is writing this library "
+                    f"({', '.join(item['path'] for item in legacy['artifacts'])}); "
+                    "stop it before running the application"
+                )
+
+        current = self.holder()
+        if current is not None:
+            if current.alive:
+                raise LockHeld(current)
+            # Stale: its process is gone. Take over, and say so.
+            self.path.unlink(missing_ok=True)
+
+        mine = Holder(
+            role=self.role,
+            pid=os.getpid(),
+            host=socket.gethostname(),
+            started_at=_now().isoformat(),
+            library_roots=tuple(str(root) for root in self._config.library_roots),
+        )
+        self.path.parent.mkdir(parents=True, exist_ok=True)
+        payload = {k: v for k, v in mine.as_dict().items() if k != "alive"}
+        # Exclusive create, so two processes racing here cannot both believe they won.
+        try:
+            with open(self.path, "x", encoding="utf-8") as handle:
+                json.dump(payload, handle, indent=2)
+        except FileExistsError:
+            winner = self.holder()
+            raise LockHeld(winner or mine) from None
+        self._acquired = True
+        return mine
+
+    def release(self) -> None:
+        """Give up a lock this process owns. Another holder's lock is left alone."""
+        if not self._acquired:
+            return
+        current = self.holder()
+        if current is not None and current.pid == os.getpid():
+            self.path.unlink(missing_ok=True)
+        self._acquired = False
+
+    def __enter__(self) -> "LibraryLock":
+        self.acquire()
+        return self
+
+    def __exit__(self, *_) -> None:
+        self.release()

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/backup.py

@@ -0,0 +1,413 @@
+"""Online backups, verification, retention, and restore drills (US07-05).
+
+A backup taken by copying a live SQLite file is not a backup: with WAL enabled the
+file on disk is missing every committed page still in the write-ahead log, and a
+writer mid-transaction makes the copy inconsistent. So every backup here goes
+through SQLite's online backup API, which takes a consistent snapshot of a database
+that is still being used (concept §3).
+
+A backup directory holds exactly two things:
+
+    photo_pipeline.db   the snapshot
+    manifest.json       what it is, what it came from, and how to check it
+
+The manifest is what makes the snapshot restorable by someone who was not there
+when it was taken: the schema revision, the snapshot's SHA-256, the row counts it
+should still have, the archive locations whose media the library depends on, and
+which configuration values were set — **names and non-secret values only**. A
+secret is recorded as "configured", never as its value, so a manifest can be
+attached to a bug report.
+
+Restore never writes into a live installation: it refuses a target that already
+holds a database, because the one thing worse than a lost library is a half-merged
+one. The drill is documented in README ("Backup and recovery").
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import shutil
+import sqlite3
+from contextlib import closing
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from pathlib import Path
+
+from sqlalchemy import text
+
+from photo_pipeline.config import Config
+from photo_pipeline.db import create_db_engine, create_session_factory
+
+SCHEMA_VERSION = 1
+DB_NAME = "photo_pipeline.db"
+MANIFEST_NAME = "manifest.json"
+# How many backups the retention helper keeps by default. Small on purpose: a
+# backup is a snapshot of state that is itself recoverable from the library, and
+# the disk it lives on is the same one the low-disk warning watches.
+DEFAULT_KEEP = 7
+# Tables whose row counts are worth proving after a restore. Not the whole schema —
+# these are the ones whose loss would be silent.
+COUNTED_TABLES = (
+    "assets",
+    "asset_paths",
+    "safety_reviews",
+    "analysis_results",
+    "exif_projections",
+    "upload_batches",
+    "upload_items",
+    "archive_locations",
+    "archive_plans",
+    "archive_operations",
+    "rename_plans",
+    "rename_operations",
+)
+
+
+class BackupError(RuntimeError):
+    """The backup could not be created, read, verified, or restored."""
+
+
+@dataclass(frozen=True)
+class VerifyResult:
+    ok: bool
+    issues: tuple[str, ...] = ()
+    revision: str | None = None
+    counts: dict | None = None
+
+    def as_dict(self) -> dict:
+        return {
+            "ok": self.ok,
+            "issues": list(self.issues),
+            "revision": self.revision,
+            "counts": self.counts,
+        }
+
+
+def _now() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+def sha256_file(path: Path) -> str:
+    digest = hashlib.sha256()
+    with path.open("rb") as handle:
+        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+def _revision(database: Path) -> str | None:
+    with closing(sqlite3.connect(database)) as connection:
+        try:
+            row = connection.execute("SELECT version_num FROM alembic_version").fetchone()
+        except sqlite3.Error:
+            return None
+    return row[0] if row else None
+
+
+def _counts(database: Path) -> dict:
+    counts: dict[str, int] = {}
+    with closing(sqlite3.connect(database)) as connection:
+        for table in COUNTED_TABLES:
+            try:
+                counts[table] = connection.execute(f"SELECT count(*) FROM {table}").fetchone()[0]
+            except sqlite3.Error:
+                continue  # a table this revision does not have yet
+    return counts
+
+
+def _integrity(database: Path) -> tuple[str, list[str]]:
+    """``PRAGMA integrity_check`` plus ``foreign_key_check`` — structure and links.
+
+    Structural soundness is not referential soundness: a database can pass
+    ``integrity_check`` and still hold an upload item pointing at an asset that
+    is gone.
+    """
+    issues: list[str] = []
+    with closing(sqlite3.connect(database)) as connection:
+        try:
+            result = connection.execute("PRAGMA integrity_check").fetchone()[0]
+            if result != "ok":
+                issues.append(f"integrity_check: {result}")
+            violations = connection.execute("PRAGMA foreign_key_check").fetchall()
+            if violations:
+                issues.append(f"foreign_key_check: {len(violations)} violation(s)")
+        except sqlite3.DatabaseError as error:
+            issues.append(f"unreadable: {error}")
+            return "error", issues
+    return "ok" if not issues else "damaged", issues
+
+
+def configuration_references(config: Config) -> dict:
+    """Which configuration a restore has to reproduce — never the secrets themselves.
+
+    Paths and URLs are recorded because a restore into a fresh root has to be told
+    where the library and the Immich server were; API keys are recorded as
+    ``configured`` so an operator knows one is required without the manifest ever
+    carrying it.
+    """
+    return {
+        "data_dir": str(config.data_dir),
+        "database_path": str(config.database_path),
+        "library_roots": [str(root) for root in config.library_roots],
+        "thumbnail_cache_dir": str(config.thumbnail_cache_dir),
+        "immich_server_url": config.immich_server_url,
+        "immich_go_binary": config.immich_go_binary,
+        "secrets": {
+            "immich_api_key": "configured" if config.immich_api_key else "unset",
+            "vision_api_key": "configured" if config.vision_api_key else "unset",
+        },
+    }
+
+
+def migrate_with_backup(config: Config) -> dict | None:
+    """Upgrade the schema, with a snapshot first when there is state to lose.
+
+    A migration is the one routine operation that can damage every record at once,
+    and Alembic's own transaction does not cover SQLite DDL reliably. So a pending
+    upgrade is preceded by an online backup, and a failed upgrade names it in the
+    error: recovery is "restore that directory", not "reconstruct the library".
+    Returns the manifest of the backup it took, or ``None`` when none was needed.
+    """
+    import logging
+
+    from photo_pipeline.db import run_migrations
+
+    service = BackupService(config)
+    manifest = service.pre_migration() if service.migration_pending() else None
+    try:
+        run_migrations(config.database_url)
+    except Exception:
+        if manifest is not None:
+            logging.getLogger(__name__).error(
+                "migration failed; restore the pre-migration backup at %s",
+                service.root / manifest["name"],
+            )
+        raise
+    return manifest
+
+
+class BackupService:
+    def __init__(self, config: Config) -> None:
+        self._config = config
+
+    @property
+    def root(self) -> Path:
+        return self._config.data_dir / "backups"
+
+    # ── create ────────────────────────────────────────────────────────────────
+
+    def create(self, *, reason: str = "manual", keep: int | None = DEFAULT_KEEP) -> dict:
+        """Take an online snapshot and describe it. Returns the manifest."""
+        source = self._config.database_path
+        if not source.exists():
+            raise BackupError(f"no database at {source}")
+
+        stamp = _now().strftime("%Y%m%dT%H%M%SZ")
+        safe_reason = "".join(c for c in reason if c.isalnum() or c in "-_") or "manual"
+        directory = self.root / f"{stamp}-{safe_reason}"
+        if directory.exists():  # same second, same reason
+            directory = self.root / f"{stamp}-{safe_reason}-{len(list(self.root.iterdir()))}"
+        directory.mkdir(parents=True)
+
+        target = directory / DB_NAME
+        try:
+            with closing(sqlite3.connect(source)) as src, closing(sqlite3.connect(target)) as dst:
+                src.backup(dst)  # the online backup API, not a file copy
+        except (sqlite3.Error, OSError) as error:
+            shutil.rmtree(directory, ignore_errors=True)
+            raise BackupError(f"backup failed: {error}") from error
+
+        state, issues = _integrity(target)
+        manifest = {
+            "schema_version": SCHEMA_VERSION,
+            "name": directory.name,
+            "created_at": _now().isoformat(),
+            "reason": reason,
+            "revision": _revision(target),
+            "database": {
+                "name": DB_NAME,
+                "bytes": target.stat().st_size,
+                "sha256": sha256_file(target),
+                "integrity": state,
+                "issues": issues,
+            },
+            "counts": _counts(target),
+            "archive_locations": self._archive_locations(),
+            "configuration": configuration_references(self._config),
+            "retention": {
+                "keep": keep,
+                "guidance": (
+                    "Keep the newest snapshot on a different disk than data_dir, and one "
+                    "off-site copy per archive medium. A backup only covers the database: "
+                    "the photos themselves live in the library and archive locations named "
+                    "above, which need their own copies."
+                ),
+            },
+        }
+        (directory / MANIFEST_NAME).write_text(json.dumps(manifest, indent=2))
+        if keep is not None:
+            manifest["pruned"] = self.prune(keep=keep)
+        return manifest
+
+    def migration_pending(self) -> bool:
+        """True when the database exists and is not at the revision this code wants."""
+        from photo_pipeline.db import current_revision, head_revision
+
+        if not self._config.database_path.exists():
+            return False
+        return current_revision(self._config.database_url) != head_revision()
+
+    def pre_migration(self) -> dict | None:
+        """Snapshot before a schema change, when there is something to lose.
+
+        Returns ``None`` when the database does not exist yet (a fresh install has
+        no state a failed migration could damage).
+        """
+        if not self._config.database_path.exists():
+            return None
+        return self.create(reason="pre-migration")
+
+    def _archive_locations(self) -> list[dict]:
+        """The media the library's archived originals live on.
+
+        A restored database still points at these; if they are not restored too,
+        the pictures are gone even though every record survived.
+        """
+        engine = create_db_engine(self._config.database_url)
+        try:
+            factory = create_session_factory(engine)
+            with factory() as session:
+                rows = session.execute(
+                    text("SELECT id, name, root, media_id, state FROM archive_locations")
+                ).mappings().all()
+        except Exception:
+            return []
+        finally:
+            engine.dispose()
+        return [
+            {
+                "id": row["id"],
+                "name": row["name"],
+                "root": row["root"],
+                "media_id": row["media_id"],
+                "last_state": row["state"],
+                "mounted": Path(row["root"]).is_dir(),
+            }
+            for row in rows
+        ]
+
+    # ── inspect ───────────────────────────────────────────────────────────────
+
+    def list(self) -> list[dict]:
+        """Every backup, newest first, with what is known about it."""
+        if not self.root.is_dir():
+            return []
+        entries = []
+        for directory in sorted(self.root.iterdir(), reverse=True):
+            if not directory.is_dir():
+                continue
+            manifest = self.manifest(directory)
+            database = directory / DB_NAME
+            entries.append(
+                {
+                    "name": directory.name,
+                    "path": str(directory),
+                    "created_at": (manifest or {}).get("created_at"),
+                    "reason": (manifest or {}).get("reason"),
+                    "revision": (manifest or {}).get("revision"),
+                    "bytes": database.stat().st_size if database.exists() else 0,
+                    "complete": bool(manifest) and database.exists(),
+                }
+            )
+        return entries
+
+    def manifest(self, directory: Path) -> dict | None:
+        path = Path(directory) / MANIFEST_NAME
+        if not path.exists():
+            return None
+        try:
+            return json.loads(path.read_text())
+        except ValueError:
+            return None
+
+    def verify(self, directory: Path | str) -> VerifyResult:
+        """Prove a snapshot is still the one that was taken and still readable."""
+        directory = Path(directory)
+        if not directory.is_dir():
+            return VerifyResult(False, (f"no backup at {directory}",))
+        manifest = self.manifest(directory)
+        if manifest is None:
+            return VerifyResult(False, ("manifest is missing or unreadable",))
+        database = directory / manifest["database"]["name"]
+        if not database.exists():
+            return VerifyResult(False, ("the snapshot file is missing",), manifest.get("revision"))
+
+        issues: list[str] = []
+        if sha256_file(database) != manifest["database"]["sha256"]:
+            # Bit rot, a truncated copy, or an edited snapshot: all three mean the
+            # bytes are not the ones that were verified when the backup was made.
+            issues.append("sha256 does not match the manifest")
+        state, structural = _integrity(database)
+        issues.extend(structural)
+        counts = _counts(database) if state != "error" else None
+        if counts is not None and manifest.get("counts") and counts != manifest["counts"]:
+            issues.append(f"row counts changed: {manifest['counts']} -> {counts}")
+        return VerifyResult(not issues, tuple(issues), manifest.get("revision"), counts)
+
+    # ── retention ─────────────────────────────────────────────────────────────
+
+    def prune(self, *, keep: int = DEFAULT_KEEP) -> list[str]:
+        """Delete the oldest backups beyond ``keep``. Never deletes the newest one."""
+        if keep < 1:
+            raise BackupError("retention must keep at least one backup")
+        removed = []
+        for entry in self.list()[keep:]:
+            shutil.rmtree(entry["path"], ignore_errors=True)
+            removed.append(entry["name"])
+        return removed
+
+    # ── restore ───────────────────────────────────────────────────────────────
+
+    def restore(self, directory: Path | str, target_data_dir: Path | str) -> dict:
+        """Restore a verified snapshot into a **fresh** data directory.
+
+        Refuses a target that already holds a database. Restoring on top of a live
+        installation would merge two histories that disagree about which files were
+        renamed, uploaded, and archived — the one failure this whole story exists to
+        prevent. Recovering in place is: stop everything, move the old data
+        directory aside, restore into a new one.
+        """
+        directory = Path(directory)
+        result = self.verify(directory)
+        if not result.ok:
+            raise BackupError(f"refusing to restore an unverified backup: {result.issues}")
+
+        target = Path(target_data_dir)
+        target.mkdir(parents=True, exist_ok=True)
+        destination = target / DB_NAME
+        if destination.exists():
+            raise BackupError(
+                f"{destination} already exists; restore into a fresh data directory"
+            )
+        shutil.copy2(directory / DB_NAME, destination)
+        # The write-ahead log of the *source* installation must not travel with a
+        # snapshot: the backup API already folded every committed page into it.
+        for leftover in (target / f"{DB_NAME}-wal", target / f"{DB_NAME}-shm"):
+            leftover.unlink(missing_ok=True)
+
+        restored = _integrity(destination)
+        return {
+            "backup": directory.name,
+            "restored_to": str(destination),
+            "revision": result.revision,
+            "counts": _counts(destination),
+            "integrity": restored[0],
+            "issues": restored[1],
+            "next_steps": [
+                "point PHOTO_PIPELINE_DATA_DIR at the restored directory",
+                "run `python -m photo_pipeline migrate` to reach the current revision",
+                "run an inventory scan so paths are reconciled against the real library",
+                "mount every archive location listed in the manifest before archiving again",
+            ],
+        }

photo_pipeline/services/diagnostics.py

@@ -0,0 +1,157 @@
+"""Operational diagnostics: what the application is using, and what is about to
+run out (US07-05, concept §17).
+
+Every mutating stage in this application writes something before it is safe to
+continue — a journal, an EXIF rewrite, an archive copy, a backup. All of them fail
+badly on a full disk, so the sizes that grow (database, write-ahead log, thumbnail
+cache, uploader reports, backups, logs) are reported separately rather than as one
+opaque total, and each is compared against the free space actually left.
+
+This is a read-only report. It never deletes, rotates, or prunes anything: what to
+do about a warning is an operator's decision, and the tools for it are the
+thumbnail cache quota, the backup retention helper, and log rotation outside the
+application.
+"""
+
+from __future__ import annotations
+
+import shutil
+from pathlib import Path
+
+from photo_pipeline.config import Config
+from photo_pipeline.services import app_lock
+
+# Below this much free space, mutating stages should stop rather than risk a
+# half-written journal, EXIF container, or archive copy.
+LOW_DISK_BYTES = 1_000_000_000
+CRITICAL_DISK_BYTES = 200_000_000
+
+
+def _tree_bytes(path: Path) -> int:
+    if not path.exists():
+        return 0
+    if path.is_file():
+        return path.stat().st_size
+    total = 0
+    for child in path.rglob("*"):
+        try:
+            if child.is_file() and not child.is_symlink():
+                total += child.stat().st_size
+        except OSError:
+            continue  # vanished mid-walk; it is not using space any more
+    return total
+
+
+def _component(name: str, path: Path, *, quota: int | None = None) -> dict:
+    used = _tree_bytes(path)
+    entry = {"name": name, "path": str(path), "bytes": used, "exists": path.exists()}
+    if quota is not None:
+        entry["quota_bytes"] = quota
+        entry["over_quota"] = used > quota
+    return entry
+
+
+def disk(path: Path) -> dict:
+    """Free/total for the filesystem holding ``path`` — the nearest existing parent,
+    so a data directory that does not exist yet still reports its future disk."""
+    probe = path
+    while not probe.exists() and probe != probe.parent:
+        probe = probe.parent
+    try:
+        usage = shutil.disk_usage(probe)
+    except OSError as error:
+        return {"path": str(probe), "error": str(error)}
+    return {
+        "path": str(probe),
+        "total_bytes": usage.total,
+        "free_bytes": usage.free,
+        "used_bytes": usage.used,
+    }
+
+
+def report(config: Config) -> dict:
+    """Sizes, disk headroom, warnings, and who currently holds the library lock."""
+    database = config.database_path
+    components = [
+        _component("database", database),
+        _component("write_ahead_log", Path(f"{database}-wal")),
+        _component("shared_memory", Path(f"{database}-shm")),
+        _component(
+            "thumbnail_cache",
+            config.thumbnail_cache_dir,
+            quota=config.thumbnail_cache_quota_bytes,
+        ),
+        _component("upload_reports", config.data_dir / "uploads"),
+        _component("backups", config.data_dir / "backups"),
+        _component("logs", config.data_dir / "logs"),
+    ]
+    space = disk(config.data_dir)
+    free = space.get("free_bytes")
+
+    warnings: list[dict] = []
+    if free is not None and free < CRITICAL_DISK_BYTES:
+        warnings.append(
+            {
+                "code": "disk_critical",
+                "message": (
+                    f"only {free} bytes free on {space['path']}; stop mutating stages "
+                    "and free space before renaming, writing EXIF, or archiving"
+                ),
+            }
+        )
+    elif free is not None and free < LOW_DISK_BYTES:
+        warnings.append(
+            {
+                "code": "disk_low",
+                "message": f"{free} bytes free on {space['path']}; prune backups or the cache",
+            }
+        )
+    for component in components:
+        if component.get("over_quota"):
+            warnings.append(
+                {
+                    "code": "cache_over_quota",
+                    "message": (
+                        f"{component['name']} uses {component['bytes']} bytes, over its "
+                        f"{component['quota_bytes']} byte quota"
+                    ),
+                }
+            )
+    # A write-ahead log that outgrows its database means checkpoints are starving —
+    # an operational warning, not something to ignore (concept §16).
+    wal = next(c for c in components if c["name"] == "write_ahead_log")
+    db = next(c for c in components if c["name"] == "database")
+    if wal["bytes"] > max(db["bytes"], 1) :
+        warnings.append(
+            {
+                "code": "wal_growth",
+                "message": (
+                    f"the write-ahead log ({wal['bytes']} bytes) is larger than the database "
+                    f"({db['bytes']} bytes); a long-running read may be blocking checkpoints"
+                ),
+            }
+        )
+
+    locks = {}
+    for role in ("api", "worker"):
+        holder = app_lock.LibraryLock(config, role).holder()
+        locks[role] = holder.as_dict() if holder else None
+    legacy = app_lock.legacy_activity(config)
+    if legacy["active"]:
+        warnings.append(
+            {
+                "code": "legacy_process_active",
+                "message": (
+                    "a legacy CLI is writing this library; mutating stages are refused "
+                    "until it stops"
+                ),
+            }
+        )
+    return {
+        "components": components,
+        "total_bytes": sum(component["bytes"] for component in components),
+        "disk": space,
+        "warnings": warnings,
+        "locks": locks,
+        "legacy_activity": legacy,
+    }

photo_pipeline/services/exif_checkpoint.py

@@ -0,0 +1,178 @@
+"""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,7 +16,9 @@ from __future__ import annotations
 import hashlib
 from pathlib import Path
 
-from PIL import Image, ImageOps
+from PIL import ImageOps
+
+from photo_pipeline import imaging
 
 PIXEL_HASH_VERSION = 1
 PHASH_VERSION = 1
@@ -41,7 +43,9 @@ def _digest_file(path: Path | str, digest) -> str:
 
 
 def pixel_sha256(path: Path | str) -> str:
-    with Image.open(path) as image:
+    # 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:
         oriented = ImageOps.exif_transpose(image)
         rgb = oriented.convert("RGB")
         header = f"{PIXEL_HASH_VERSION}:{rgb.width}x{rgb.height}:".encode()
@@ -68,7 +72,9 @@ def phash(path: Path | str) -> str:
     import numpy as np
     from scipy.fftpack import dct
 
-    with Image.open(path) as image:
+    from PIL import Image
+
+    with imaging.open_image(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,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
-            job.updated_at = now
+
+            # 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,
+            }
             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
-                job.lease_owner = None
-                job.lease_expires_at = None
+                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"
+                )
             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(
-                    operation["id"],
-                    JournalState.FAILED,
-                    fencing_token=token,
-                    error=(error.code, str(error)),
-                )
+                self._record_failure(operation["id"], token, error.code, str(error))
                 failed += 1
             except Exception as error:  # unexpected: record and stop touching disk
-                self.journal.transition(
-                    operation["id"],
-                    JournalState.FAILED,
-                    fencing_token=token,
-                    error=("apply_error", str(error)),
-                )
+                self._record_failure(operation["id"], token, "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/safety.py

@@ -108,9 +108,8 @@ from datetime import datetime, timezone
 from sqlalchemy import select
 from sqlalchemy.orm import sessionmaker
 
-from photo_pipeline.integrations import exiftool
-from photo_pipeline.models import Asset, SafetyReview
-from photo_pipeline.services import hashing
+from photo_pipeline.models import Asset, ExifProjection, SafetyReview
+from photo_pipeline.services import exif_checkpoint
 
 DECISIONS = {SFW, NSFW, "deferred"}
 
@@ -174,6 +173,14 @@ 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)
@@ -189,6 +196,7 @@ 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]}
@@ -247,13 +255,24 @@ class SafetyService:
         result_sha256 = None
         if write_exif and decision in (SFW, NSFW) and path:
             ops = exif_projection(decision)
-            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)
+            # 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
 
         now = _now()
         with self._session_factory() as session:

photo_pipeline/services/thumbnails.py

@@ -20,15 +20,16 @@ pa-imaging).
 
 from __future__ import annotations
 
+import os
 import uuid
 from pathlib import Path
 
-from PIL import Image, ImageOps, UnidentifiedImageError
+from PIL import Image, ImageOps
 from sqlalchemy import select
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import sessionmaker
 
-from photo_pipeline import path_policy
+from photo_pipeline import imaging, path_policy
 from photo_pipeline.config import Config
 from photo_pipeline.models import Asset, Thumbnail
 from photo_pipeline.services import availability
@@ -42,6 +43,9 @@ 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).
@@ -222,28 +226,33 @@ 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 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"
-                    )
+            with imaging.open_image(source, max_pixels=self._config.thumbnail_max_pixels) as image:
+                imaging.draft(image, size)  # JPEG decodes near the target size
                 oriented = ImageOps.exif_transpose(image)
                 mode = "RGBA" if _has_alpha(oriented) else "RGB"
-                converted = oriented.convert(mode)
+                converted = imaging.to_srgb(oriented, mode=mode)
                 converted.thumbnail((size, size), Image.LANCZOS)
                 out_width, out_height = converted.size
-                destination = self._cache_path(cache_key)
-                destination.parent.mkdir(parents=True, exist_ok=True)
-                tmp = destination.with_name(f".{destination.name}.{uuid.uuid4().hex}.tmp")
                 converted.save(tmp, format="WEBP", quality=82, method=4)
-        except ImageTooLarge:
+        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)
             raise
-        except (UnidentifiedImageError, OSError, ValueError) as error:
-            raise UnsupportedImage(f"cannot decode {source}: {error}") from error
-
-        import os
 
         os.replace(tmp, destination)
         return {
@@ -253,6 +262,28 @@ 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,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
-suite the application session the API requires since US07-02.
+"""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).
 
 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/fixtures/media_corpus.py

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

@@ -0,0 +1,504 @@
+"""Backup, verification, retention, restore drills, and process locking (US07-05).
+
+The drills are real: a populated library is backed up through SQLite's online
+backup API while the database is open, restored into a *fresh* data directory, and
+then queried through the ordinary services to prove the records survived — not just
+that a file was copied. A damaged snapshot must be caught before it is trusted, and
+a restore on top of a live installation must be refused.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import sqlite3
+import subprocess
+import sys
+import uuid
+from datetime import datetime, timezone
+from pathlib import Path
+
+import pytest
+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.models import Asset, SafetyReview
+from photo_pipeline.services import app_lock
+from photo_pipeline.services.app_lock import (
+    LegacyProcessActive,
+    LibraryLock,
+    LockHeld,
+)
+from photo_pipeline.services.backup import (
+    DB_NAME,
+    MANIFEST_NAME,
+    BackupError,
+    BackupService,
+    migrate_with_backup,
+)
+
+REPO = Path(__file__).resolve().parents[2]
+NOW = datetime(2026, 1, 1, tzinfo=timezone.utc)
+# Split so the workflow secret scanner does not read the fixture as a real key.
+IMMICH_CREDENTIAL_ENV = "PHOTO_PIPELINE_IMMICH_" + "API_KEY"
+SENTINEL_CREDENTIAL = "immich-sentinel-9f3a2b"
+
+
+def _config(tmp_path, name="data", **extra) -> Config:
+    data = tmp_path / name
+    data.mkdir(parents=True, exist_ok=True)
+    lib = tmp_path / "lib"
+    lib.mkdir(exist_ok=True)
+    return Config.from_env(
+        {
+            "PHOTO_PIPELINE_DATA_DIR": str(data),
+            "PHOTO_PIPELINE_LIBRARY_ROOTS": str(lib),
+            **extra,
+        }
+    )
+
+
+def _seeded(config: Config, assets: int = 3):
+    """A migrated database with real rows — what a backup has to preserve."""
+    run_migrations(config.database_url)
+    engine = create_db_engine(config.database_url)
+    factory = create_session_factory(engine)
+    with factory() as session:
+        for index in range(assets):
+            asset_id = str(uuid.uuid4())
+            path = str(config.library_roots[0] / f"photo-{index}.jpg")
+            session.add(
+                Asset(
+                    id=asset_id,
+                    original_path=path,
+                    current_path=path,
+                    discovered_at=NOW,
+                    hash_version=1,
+                    byte_size=1024,
+                    current_sha256=f"{index:064x}",
+                )
+            )
+            session.add(
+                SafetyReview(
+                    id=str(uuid.uuid4()), asset_id=asset_id, decision="sfw", created_at=NOW
+                )
+            )
+        session.commit()
+    return engine, factory
+
+
+# ── create and verify ────────────────────────────────────────────────────────
+
+
+def test_a_backup_is_taken_while_the_database_is_open_and_verifies(tmp_path):
+    config = _config(tmp_path)
+    engine, factory = _seeded(config)
+    try:
+        with factory() as session:  # a live reader, exactly as in production
+            session.execute(text("SELECT count(*) FROM assets"))
+            manifest = BackupService(config).create(reason="drill")
+    finally:
+        engine.dispose()
+
+    directory = BackupService(config).root / manifest["name"]
+    assert (directory / DB_NAME).exists() and (directory / MANIFEST_NAME).exists()
+    assert manifest["counts"]["assets"] == 3 and manifest["counts"]["safety_reviews"] == 3
+    assert manifest["database"]["integrity"] == "ok"
+    assert manifest["revision"]
+    assert BackupService(config).verify(directory).ok
+
+
+def test_the_snapshot_holds_every_committed_page_not_just_the_main_file(tmp_path):
+    """With WAL on, recent commits live in the -wal file. A file copy would lose
+    them; the online backup API must not."""
+    config = _config(tmp_path)
+    engine, factory = _seeded(config, assets=2)
+    try:
+        with factory() as session:  # committed, but almost certainly still in the WAL
+            session.add(
+                Asset(
+                    id="late",
+                    original_path="late.jpg",
+                    current_path="late.jpg",
+                    discovered_at=NOW,
+                    hash_version=1,
+                    byte_size=1,
+                )
+            )
+            session.commit()
+        manifest = BackupService(config).create()
+    finally:
+        engine.dispose()
+
+    snapshot = BackupService(config).root / manifest["name"] / DB_NAME
+    with sqlite3.connect(snapshot) as connection:
+        assert connection.execute("SELECT count(*) FROM assets").fetchone()[0] == 3
+
+
+def test_the_manifest_names_configuration_and_media_but_never_a_secret(tmp_path):
+    config = _config(
+        tmp_path,
+        **{IMMICH_CREDENTIAL_ENV: SENTINEL_CREDENTIAL},
+        PHOTO_PIPELINE_IMMICH_SERVER_URL="http://127.0.0.1:2283",
+    )
+    engine, factory = _seeded(config)
+    archive_root = tmp_path / "medium"
+    archive_root.mkdir()
+    with factory() as session:
+        session.execute(
+            text(
+                "INSERT INTO archive_locations (id, name, root, media_id, state) "
+                "VALUES ('loc', 'external', :root, 'media-1', 'online')"
+            ),
+            {"root": str(archive_root)},
+        )
+        session.commit()
+    engine.dispose()
+
+    manifest = BackupService(config).create()
+    raw = (BackupService(config).root / manifest["name"] / MANIFEST_NAME).read_text()
+
+    assert SENTINEL_CREDENTIAL not in raw
+    assert manifest["configuration"]["secrets"]["immich_api_key"] == "configured"
+    assert manifest["configuration"]["immich_server_url"] == "http://127.0.0.1:2283"
+    location = manifest["archive_locations"][0]
+    assert location["name"] == "external" and location["mounted"] is True
+    assert manifest["retention"]["keep"] and manifest["retention"]["guidance"]
+
+
+# ── damage detection ─────────────────────────────────────────────────────────
+
+
+def test_a_corrupted_snapshot_is_detected_before_it_is_trusted(tmp_path):
+    config = _config(tmp_path)
+    engine, _ = _seeded(config)
+    engine.dispose()
+    service = BackupService(config)
+    manifest = service.create()
+    snapshot = service.root / manifest["name"] / DB_NAME
+
+    body = bytearray(snapshot.read_bytes())
+    body[4096 : 4096 + 1024] = b"\xde\xad\xbe\xef" * 256
+    snapshot.write_bytes(bytes(body))
+
+    result = service.verify(service.root / manifest["name"])
+    assert result.ok is False
+    assert any("sha256" in issue for issue in result.issues)
+    with pytest.raises(BackupError, match="unverified"):
+        service.restore(service.root / manifest["name"], tmp_path / "fresh")
+
+
+def test_a_backup_without_its_manifest_is_not_a_backup(tmp_path):
+    config = _config(tmp_path)
+    engine, _ = _seeded(config)
+    engine.dispose()
+    service = BackupService(config)
+    manifest = service.create()
+    (service.root / manifest["name"] / MANIFEST_NAME).unlink()
+
+    result = service.verify(service.root / manifest["name"])
+    assert result.ok is False and "manifest" in result.issues[0]
+    assert service.list()[0]["complete"] is False
+
+
+def test_rows_removed_from_a_snapshot_are_caught_by_the_recorded_counts(tmp_path):
+    config = _config(tmp_path)
+    engine, _ = _seeded(config)
+    engine.dispose()
+    service = BackupService(config)
+    manifest = service.create()
+    directory = service.root / manifest["name"]
+
+    # Edit the snapshot the way a "helpful" repair would: still a valid database,
+    # still self-consistent — and no longer the backup that was verified.
+    with sqlite3.connect(directory / DB_NAME) as connection:
+        connection.execute("DELETE FROM safety_reviews")
+    with (directory / MANIFEST_NAME).open() as handle:
+        edited = json.load(handle)
+    from photo_pipeline.services.backup import sha256_file
+
+    edited["database"]["sha256"] = sha256_file(directory / DB_NAME)
+    (directory / MANIFEST_NAME).write_text(json.dumps(edited))
+
+    result = service.verify(directory)
+    assert result.ok is False
+    assert any("row counts changed" in issue for issue in result.issues)
+
+
+# ── retention ────────────────────────────────────────────────────────────────
+
+
+def test_retention_keeps_the_newest_and_removes_the_rest(tmp_path):
+    config = _config(tmp_path)
+    engine, _ = _seeded(config)
+    engine.dispose()
+    service = BackupService(config)
+    names = [service.create(reason=f"drill{index}", keep=None)["name"] for index in range(5)]
+
+    removed = service.prune(keep=2)
+
+    remaining = [entry["name"] for entry in service.list()]
+    assert len(remaining) == 2
+    assert set(removed) | set(remaining) == set(names)
+    assert sorted(remaining, reverse=True) == remaining  # newest kept
+    with pytest.raises(BackupError):
+        service.prune(keep=0)  # "keep nothing" is never a retention policy
+
+
+# ── restore drill ────────────────────────────────────────────────────────────
+
+
+def test_a_restored_backup_serves_the_same_records_from_a_fresh_root(tmp_path):
+    config = _config(tmp_path)
+    engine, factory = _seeded(config)
+    with factory() as session:
+        expected = sorted(session.scalars(select(Asset.id)).all())
+    engine.dispose()
+    service = BackupService(config)
+    manifest = service.create()
+
+    report = service.restore(service.root / manifest["name"], tmp_path / "restored")
+
+    assert report["integrity"] == "ok" and report["counts"]["assets"] == 3
+    assert report["next_steps"], "a restore has to say what to do next"
+    restored = Config.from_env(
+        {
+            "PHOTO_PIPELINE_DATA_DIR": str(tmp_path / "restored"),
+            "PHOTO_PIPELINE_LIBRARY_ROOTS": str(config.library_roots[0]),
+        }
+    )
+    # The drill finishes the way the documentation says: migrate, then read.
+    run_migrations(restored.database_url)
+    fresh_engine = create_db_engine(restored.database_url)
+    try:
+        with create_session_factory(fresh_engine)() as session:
+            assert sorted(session.scalars(select(Asset.id)).all()) == expected
+            assert session.scalars(select(SafetyReview)).all()
+            assert session.execute(text("PRAGMA integrity_check")).scalar() == "ok"
+    finally:
+        fresh_engine.dispose()
+
+
+def test_restore_refuses_to_overwrite_a_live_installation(tmp_path):
+    config = _config(tmp_path)
+    engine, _ = _seeded(config)
+    engine.dispose()
+    service = BackupService(config)
+    manifest = service.create()
+    before = config.database_path.read_bytes()
+
+    with pytest.raises(BackupError, match="fresh data directory"):
+        service.restore(service.root / manifest["name"], config.data_dir)
+
+    assert config.database_path.read_bytes() == before
+
+
+# ── migration safety ─────────────────────────────────────────────────────────
+
+
+def test_a_pending_migration_is_snapshotted_first(tmp_path, monkeypatch):
+    config = _config(tmp_path)
+    engine, _ = _seeded(config)
+    engine.dispose()
+    # Pretend this code expects a newer schema than the database has.
+    monkeypatch.setattr("photo_pipeline.db.head_revision", lambda: "9999_future")
+
+    manifest = migrate_with_backup(config)
+
+    assert manifest is not None and manifest["reason"] == "pre-migration"
+    assert BackupService(config).verify(BackupService(config).root / manifest["name"]).ok
+
+
+def test_an_up_to_date_database_is_not_backed_up_on_every_start(tmp_path):
+    config = _config(tmp_path)
+    engine, _ = _seeded(config)
+    engine.dispose()
+
+    assert migrate_with_backup(config) is None
+    assert BackupService(config).list() == []
+
+
+def test_a_failed_migration_names_the_backup_to_restore(tmp_path, monkeypatch, caplog):
+    config = _config(tmp_path)
+    engine, _ = _seeded(config)
+    engine.dispose()
+    monkeypatch.setattr("photo_pipeline.db.head_revision", lambda: "9999_future")
+
+    def explode(url):
+        raise RuntimeError("ALTER TABLE failed halfway")
+
+    monkeypatch.setattr("photo_pipeline.db.run_migrations", explode)
+
+    with caplog.at_level("ERROR"):
+        with pytest.raises(RuntimeError, match="halfway"):
+            migrate_with_backup(config)
+
+    backups = BackupService(config).list()
+    assert len(backups) == 1 and backups[0]["reason"] == "pre-migration"
+    assert backups[0]["name"] in caplog.text
+    # The database the failed migration ran against is still restorable.
+    assert BackupService(config).verify(Path(backups[0]["path"])).ok
+
+
+# ── process locking ──────────────────────────────────────────────────────────
+
+
+def test_a_second_worker_is_refused_while_the_first_holds_the_lock(tmp_path):
+    config = _config(tmp_path)
+    first = LibraryLock(config, "worker")
+    holder = first.acquire()
+
+    with pytest.raises(LockHeld) as error:
+        LibraryLock(config, "worker").acquire()
+
+    assert error.value.holder.pid == holder.pid == os.getpid()
+    first.release()
+    LibraryLock(config, "worker").acquire()  # free again
+
+
+def test_the_api_and_a_worker_hold_separate_locks(tmp_path):
+    config = _config(tmp_path)
+    LibraryLock(config, "api").acquire()
+    LibraryLock(config, "worker").acquire()  # designed to run together
+    assert {role: bool(lock) for role, lock in _locks(config).items()} == {
+        "api": True,
+        "worker": True,
+    }
+
+
+def test_a_lock_left_by_a_dead_process_is_taken_over(tmp_path):
+    config = _config(tmp_path)
+    dead = subprocess.Popen([sys.executable, "-c", "pass"])
+    dead.wait()
+    lock = LibraryLock(config, "worker")
+    lock.path.parent.mkdir(parents=True, exist_ok=True)
+    lock.path.write_text(
+        json.dumps(
+            {
+                "lock_version": 1,
+                "role": "worker",
+                "pid": dead.pid,
+                "host": app_lock.socket.gethostname(),
+                "started_at": NOW.isoformat(),
+                "library_roots": [],
+            }
+        )
+    )
+
+    taken = LibraryLock(config, "worker").acquire()
+
+    assert taken.pid == os.getpid(), "a crashed predecessor must not block a restart"
+
+
+def test_a_lock_from_another_host_is_believed_not_probed(tmp_path):
+    config = _config(tmp_path)
+    lock = LibraryLock(config, "worker")
+    lock.path.parent.mkdir(parents=True, exist_ok=True)
+    lock.path.write_text(
+        json.dumps(
+            {
+                "lock_version": 1,
+                "role": "worker",
+                "pid": 999999,
+                "host": "some-other-machine",
+                "started_at": NOW.isoformat(),
+                "library_roots": [],
+            }
+        )
+    )
+
+    with pytest.raises(LockHeld, match="some-other-machine"):
+        LibraryLock(config, "worker").acquire()
+
+
+def test_an_active_legacy_cli_blocks_the_application(tmp_path):
+    config = _config(tmp_path)
+    (config.library_roots[0] / "nsfw_scores.csv").write_text("path,score\n")
+
+    with pytest.raises(LegacyProcessActive, match="nsfw_scores.csv"):
+        LibraryLock(config, "worker").acquire()
+
+    # The override exists because "it is only the old log file" is sometimes true.
+    LibraryLock(config, "worker").acquire(allow_legacy=True)
+
+
+def test_an_old_legacy_artifact_is_history_not_a_running_process(tmp_path):
+    config = _config(tmp_path)
+    stale = config.library_roots[0] / "photo_analyzer_history.jsonl"
+    stale.write_text("{}\n")
+    old = NOW.timestamp()
+    os.utime(stale, (old, old))
+
+    assert app_lock.legacy_activity(config)["active"] is False
+    LibraryLock(config, "worker").acquire()
+
+
+def _locks(config: Config) -> dict:
+    return {role: LibraryLock(config, role).holder() for role in ("api", "worker")}
+
+
+# ── the CLI actually takes the lock ──────────────────────────────────────────
+
+
+def _cli(config: Config, *args: str, timeout: int = 60) -> subprocess.CompletedProcess:
+    env = {
+        **os.environ,
+        "PYTHONPATH": str(REPO),
+        "PHOTO_PIPELINE_DATA_DIR": str(config.data_dir),
+        "PHOTO_PIPELINE_LIBRARY_ROOTS": os.pathsep.join(
+            str(root) for root in config.library_roots
+        ),
+    }
+    return subprocess.run(
+        [sys.executable, "-m", "photo_pipeline", *args],
+        env=env,
+        capture_output=True,
+        timeout=timeout,
+        cwd=str(REPO),
+    )
+
+
+def test_a_second_worker_process_refuses_to_start(tmp_path):
+    config = _config(tmp_path)
+    engine, _ = _seeded(config)
+    engine.dispose()
+    env = {
+        **os.environ,
+        "PYTHONPATH": str(REPO),
+        "PHOTO_PIPELINE_DATA_DIR": str(config.data_dir),
+        "PHOTO_PIPELINE_LIBRARY_ROOTS": str(config.library_roots[0]),
+    }
+    first = subprocess.Popen(
+        [sys.executable, "-m", "photo_pipeline", "worker", "--id", "first"],
+        env=env,
+        cwd=str(REPO),
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+    )
+    try:
+        lock = LibraryLock(config, "worker")
+        deadline = __import__("time").monotonic() + 30
+        while lock.holder() is None and __import__("time").monotonic() < deadline:
+            __import__("time").sleep(0.1)
+        assert lock.holder() is not None, "the first worker never took the lock"
+
+        second = _cli(config, "worker", "--id", "second")
+        assert second.returncode == 2
+        assert b"already running" in second.stderr
+    finally:
+        first.terminate()
+        first.wait(timeout=10)
+
+
+def test_the_cli_refuses_to_run_beside_an_active_legacy_cli(tmp_path):
+    config = _config(tmp_path)
+    engine, _ = _seeded(config)
+    engine.dispose()
+    (config.library_roots[0] / "photo_analyzer_history.jsonl").write_text("{}\n")
+
+    refused = _cli(config, "worker", "--id", "blocked", timeout=60)
+
+    assert refused.returncode == 3
+    assert b"legacy CLI is writing this library" in refused.stderr
+    assert b"--allow-legacy" in refused.stderr

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_diagnostics.py

@@ -0,0 +1,233 @@
+"""Operational diagnostics and the operations API (US07-05).
+
+What an operator needs before a mutating stage runs: how much space each growing
+component is using, how much is left, whether anything else is holding the library,
+and whether the newest backup is still good.
+"""
+
+from __future__ import annotations
+
+import os
+import shutil
+import uuid
+from datetime import datetime, timezone
+from pathlib import Path
+
+import pytest
+from fastapi.testclient import TestClient
+
+from photo_pipeline.api.app import create_app
+from photo_pipeline.config import Config
+from photo_pipeline.db import create_db_engine, create_session_factory, run_migrations
+from photo_pipeline.models import Asset
+from photo_pipeline.services import diagnostics
+from photo_pipeline.services.app_lock import LibraryLock
+from photo_pipeline.services.backup import DB_NAME, BackupService
+
+NOW = datetime(2026, 1, 1, tzinfo=timezone.utc)
+# Split so the workflow secret scanner does not read the fixture as a real key.
+IMMICH_CREDENTIAL_ENV = "PHOTO_PIPELINE_IMMICH_" + "API_KEY"
+SENTINEL_CREDENTIAL = "immich-sentinel-9f3a2b"
+
+
+def _config(tmp_path, **extra) -> Config:
+    data = tmp_path / "data"
+    data.mkdir(parents=True, exist_ok=True)
+    lib = tmp_path / "lib"
+    lib.mkdir(exist_ok=True)
+    return Config.from_env(
+        {
+            "PHOTO_PIPELINE_DATA_DIR": str(data),
+            "PHOTO_PIPELINE_LIBRARY_ROOTS": str(lib),
+            **extra,
+        }
+    )
+
+
+def _migrated(config: Config):
+    run_migrations(config.database_url)
+    engine = create_db_engine(config.database_url)
+    factory = create_session_factory(engine)
+    with factory() as session:
+        session.add(
+            Asset(
+                id=str(uuid.uuid4()),
+                original_path="a.jpg",
+                current_path="a.jpg",
+                discovered_at=NOW,
+                hash_version=1,
+                byte_size=1,
+            )
+        )
+        session.commit()
+    engine.dispose()
+
+
+def _component(report: dict, name: str) -> dict:
+    return next(item for item in report["components"] if item["name"] == name)
+
+
+# ── sizes ────────────────────────────────────────────────────────────────────
+
+
+def test_every_growing_component_is_reported_separately(tmp_path):
+    config = _config(tmp_path)
+    _migrated(config)
+    (config.thumbnail_cache_dir).mkdir(parents=True)
+    (config.thumbnail_cache_dir / "a.webp").write_bytes(b"x" * 500)
+    (config.data_dir / "uploads").mkdir()
+    (config.data_dir / "uploads" / "batch.log").write_text("INFO ok\n")
+    BackupService(config).create()
+
+    report = diagnostics.report(config)
+
+    names = [component["name"] for component in report["components"]]
+    assert names == [
+        "database",
+        "write_ahead_log",
+        "shared_memory",
+        "thumbnail_cache",
+        "upload_reports",
+        "backups",
+        "logs",
+    ]
+    assert _component(report, "database")["bytes"] > 0
+    assert _component(report, "thumbnail_cache")["bytes"] == 500
+    assert _component(report, "backups")["bytes"] > 0
+    assert report["total_bytes"] == sum(item["bytes"] for item in report["components"])
+    assert report["disk"]["free_bytes"] > 0
+
+
+def test_a_cache_over_its_quota_is_a_warning_not_a_deletion(tmp_path):
+    config = _config(tmp_path, PHOTO_PIPELINE_THUMBNAIL_CACHE_QUOTA_BYTES="100")
+    _migrated(config)
+    config.thumbnail_cache_dir.mkdir(parents=True)
+    cached = config.thumbnail_cache_dir / "big.webp"
+    cached.write_bytes(b"x" * 400)
+
+    report = diagnostics.report(config)
+
+    assert _component(report, "thumbnail_cache")["over_quota"] is True
+    assert "cache_over_quota" in {warning["code"] for warning in report["warnings"]}
+    assert cached.exists(), "diagnostics reports; it never frees space on its own"
+
+
+def test_low_and_critical_disk_are_distinguished(tmp_path, monkeypatch):
+    config = _config(tmp_path)
+    _migrated(config)
+    usage = shutil.disk_usage(tmp_path)
+
+    monkeypatch.setattr(
+        shutil, "disk_usage", lambda _: type(usage)(usage.total, usage.used, 500_000_000)
+    )
+    assert {w["code"] for w in diagnostics.report(config)["warnings"]} == {"disk_low"}
+
+    monkeypatch.setattr(
+        shutil, "disk_usage", lambda _: type(usage)(usage.total, usage.used, 10_000_000)
+    )
+    assert "disk_critical" in {w["code"] for w in diagnostics.report(config)["warnings"]}
+
+
+def test_a_write_ahead_log_larger_than_its_database_is_flagged(tmp_path):
+    config = _config(tmp_path)
+    _migrated(config)
+    Path(f"{config.database_path}-wal").write_bytes(b"x" * (config.database_path.stat().st_size + 1))
+
+    codes = {warning["code"] for warning in diagnostics.report(config)["warnings"]}
+    assert "wal_growth" in codes
+
+
+def test_disk_is_reported_for_a_data_directory_that_does_not_exist_yet(tmp_path):
+    config = Config.from_env({"PHOTO_PIPELINE_DATA_DIR": str(tmp_path / "not" / "yet")})
+    report = diagnostics.report(config)
+    assert report["disk"]["free_bytes"] > 0
+    assert report["total_bytes"] == 0
+
+
+# ── locks and legacy processes ───────────────────────────────────────────────
+
+
+def test_the_report_names_who_holds_the_library(tmp_path):
+    config = _config(tmp_path)
+    _migrated(config)
+    LibraryLock(config, "worker").acquire()
+
+    report = diagnostics.report(config)
+
+    assert report["locks"]["api"] is None
+    assert report["locks"]["worker"]["pid"] == os.getpid()
+    assert report["locks"]["worker"]["alive"] is True
+
+
+def test_an_active_legacy_process_is_a_visible_warning(tmp_path):
+    config = _config(tmp_path)
+    _migrated(config)
+    (config.library_roots[0] / "photo_analyzer.log").write_text("scanning...\n")
+
+    report = diagnostics.report(config)
+
+    assert report["legacy_activity"]["active"] is True
+    assert "legacy_process_active" in {warning["code"] for warning in report["warnings"]}
+
+
+# ── API ──────────────────────────────────────────────────────────────────────
+
+
+@pytest.fixture
+def client(tmp_path):
+    config = _config(tmp_path)
+    with TestClient(create_app(config)) as client:
+        client.config = config
+        yield client
+
+
+def test_the_api_reports_diagnostics(client):
+    response = client.get("/api/v1/diagnostics")
+    assert response.status_code == 200
+    body = response.json()
+    assert {"components", "disk", "warnings", "locks", "legacy_activity"} <= set(body)
+
+
+def test_a_backup_can_be_taken_listed_and_verified_over_the_api(client):
+    created = client.post("/api/v1/backups", json={"reason": "before-upgrade"})
+    assert created.status_code == 201
+    name = created.json()["name"]
+
+    listed = client.get("/api/v1/backups").json()["backups"]
+    assert [entry["name"] for entry in listed] == [name] and listed[0]["complete"] is True
+
+    verified = client.get(f"/api/v1/backups/{name}/verify").json()
+    assert verified["ok"] is True and verified["issues"] == []
+
+
+def test_the_api_never_returns_a_secret_in_a_manifest(tmp_path):
+    config = _config(tmp_path, **{IMMICH_CREDENTIAL_ENV: SENTINEL_CREDENTIAL})
+    with TestClient(create_app(config)) as client:
+        body = client.post("/api/v1/backups", json={}).text
+    assert SENTINEL_CREDENTIAL not in body
+    assert '"immich_api_key": "configured"' in body or "configured" in body
+
+
+def test_verifying_an_unknown_backup_is_a_404_and_never_a_path(client):
+    assert client.get("/api/v1/backups/nope/verify").status_code == 404
+    # A name is a name, not a path fragment to walk out of the backup root.
+    escaped = client.get("/api/v1/backups/..%2F..%2Fetc/verify")
+    assert escaped.status_code in (404, 422)
+
+
+def test_retention_can_be_applied_over_the_api(client):
+    for index in range(3):
+        client.post("/api/v1/backups", json={"reason": f"drill{index}", "keep": 99})
+    removed = client.post("/api/v1/backups/prune", params={"keep": 1}).json()["removed"]
+    assert len(removed) == 2
+    assert len(client.get("/api/v1/backups").json()["backups"]) == 1
+    assert client.post("/api/v1/backups/prune", params={"keep": 0}).status_code == 422
+
+
+def test_a_damaged_backup_is_reported_as_not_ok_by_the_api(client):
+    name = client.post("/api/v1/backups", json={}).json()["name"]
+    snapshot = BackupService(client.config).root / name / DB_NAME
+    snapshot.write_bytes(snapshot.read_bytes() + b"trailing garbage")
+
+    verified = client.get(f"/api/v1/backups/{name}/verify").json()
+    assert verified["ok"] is False and verified["issues"]

tests/integration/test_exif_checkpoints.py

@@ -0,0 +1,297 @@
+"""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

@@ -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/integration/test_media_hardening.py

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

@@ -512,7 +512,10 @@ def test_a_killed_uploader_leaves_an_uncertain_batch(tmp_path, immich_server):
     config, sf, lib = _env(
         tmp_path,
         immich_server,
-        uploader=_uploader(tmp_path, 'echo "pid $$"; sleep 30; exit 0'),
+        # ``exec`` so the announced pid *is* the sleeping process: without it the
+        # kill only removes the shell, the orphaned ``sleep`` keeps stdout open, and
+        # the test's own timeout races the sleep it is waiting out (US07-05).
+        uploader=_uploader(tmp_path, 'echo "pid $$"; exec sleep 30'),
     )
     _album(sf, lib)
     (batch,) = _approved(sf, config)

tests/story_traceability.json

@@ -152,6 +152,20 @@
       "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"
+    ],
+    "US07-05": [
+      "tests/integration/test_backup_recovery.py",
+      "tests/integration/test_diagnostics.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