US07-05: Deliver Backup and Operational Recovery

README.md

@@ -281,6 +281,80 @@ file in the temporary library are copied to `.artifacts/<test id>/` before pytes
 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

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)
+        lock = LibraryLock(config, "worker")
-        engine = create_db_engine(config.database_url)
+        if (held := _acquire(lock, allow_legacy=args.allow_legacy)) is not None:
-        Worker(create_session_factory(engine), worker_id=args.id, config=config).run_forever()
+            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,7 +40,8 @@ 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
@@ -89,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
@@ -126,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:
+def _alembic_config(url: str):
-    """Upgrade the database at ``url`` to the latest revision."""
-    from alembic import command
     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/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/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,
+    }

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_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_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

@@ -162,6 +162,10 @@
       "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"
     ]
   }
 }