US07-07: Automate Full Release Acceptance (#94)

tests/e2e/test_release_gate.py

@@ -0,0 +1,298 @@
+"""The release gate, the read-only dry run, and the approval that unlocks mutation
+(US07-07).
+
+The gate itself is exercised with a tiny stage set — running the whole suite from
+inside the suite would be a fork bomb with better manners. What is proven here is
+the machinery a release depends on: the story matrix is complete, a failing stage
+fails the gate, an unexpected skip fails the gate, and the evidence is written with
+checksums that match what was written.
+
+The dry run is proven to be read-only against a real temporary library, and the
+approval is proven to be what stands between a configured library and any mutation.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+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 release
+
+REPO = Path(__file__).resolve().parents[2]
+NOW = datetime(2026, 1, 1, tzinfo=timezone.utc)
+
+# Two throwaway stages: one that passes, one the test can point at a failure.
+PASSING = ("tests/e2e/test_traceability.py",)
+
+
+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,
+        }
+    )
+
+
+# ── the story matrix ─────────────────────────────────────────────────────────
+
+
+def test_every_backlog_story_is_delivered_or_explicitly_planned():
+    matrix = release.story_matrix(REPO)
+
+    assert matrix["problems"] == [], "the story matrix has holes"
+    assert len(matrix["delivered"]) + len(matrix["planned"]) == matrix["stories"]
+    assert "US01-01" in matrix["delivered"] and "US07-07" in matrix["delivered"]
+
+
+def test_a_story_without_tests_is_a_gate_failure(tmp_path):
+    """A story file nobody covered must not pass quietly as 'no tests ran'."""
+    fake = tmp_path / "repo"
+    (fake / "delivery_backlog" / "stories").mkdir(parents=True)
+    (fake / "tests").mkdir()
+    (fake / "delivery_backlog" / "stories" / "US99-01-invented.md").write_text("# US99-01")
+    (fake / "tests" / "story_traceability.json").write_text(json.dumps({"stories": {}}))
+
+    matrix = release.story_matrix(fake)
+
+    assert matrix["problems"] == ["story with no tests and not planned: US99-01"]
+
+
+def test_a_mapping_to_a_deleted_test_file_is_a_gate_failure(tmp_path):
+    fake = tmp_path / "repo"
+    (fake / "delivery_backlog" / "stories").mkdir(parents=True)
+    (fake / "tests").mkdir()
+    (fake / "delivery_backlog" / "stories" / "US99-01-invented.md").write_text("# US99-01")
+    (fake / "tests" / "story_traceability.json").write_text(
+        json.dumps({"stories": {"US99-01": ["tests/gone.py"]}})
+    )
+
+    assert release.story_matrix(fake)["problems"] == [
+        "mapped test file is missing — US99-01: tests/gone.py"
+    ]
+
+
+# ── the gate ─────────────────────────────────────────────────────────────────
+
+
+def test_the_gate_runs_its_stages_and_keeps_checksummed_evidence(tmp_path):
+    config = _config(tmp_path)
+    evidence = tmp_path / "evidence"
+
+    report = release.run_gate(config, output=evidence, stages=(("smoke", PASSING),))
+
+    assert report["ok"] is True and report["failures"] == []
+    assert report["stages"][0]["stage"] == "smoke" and report["stages"][0]["ok"] is True
+    assert report["revision"], "the evidence must say which commit it covers"
+    assert report["matrix"]["problems"] == []
+
+    written = json.loads((evidence / release.REPORT_NAME).read_text())
+    assert written["ok"] is True
+    assert (evidence / "logs" / "smoke.log").exists()
+    checksums = (evidence / release.CHECKSUMS_NAME).read_text().splitlines()
+    assert len(checksums) >= 2
+    for line in checksums:
+        digest, name = line.split("  ", 1)
+        assert release.sha256_file(evidence / name) == digest
+
+
+def test_a_failing_stage_fails_the_gate(tmp_path):
+    config = _config(tmp_path)
+    failing = tmp_path / "failing_test.py"
+    failing.write_text("def test_no():\n    assert False\n")
+
+    report = release.run_gate(
+        config, output=tmp_path / "evidence", stages=(("broken", (str(failing),)),)
+    )
+
+    assert report["ok"] is False and report["failures"] == ["broken"]
+    assert report["stages"][0]["returncode"] != 0
+
+
+def test_an_unexpected_skip_fails_the_gate_but_an_environment_skip_does_not():
+    environment = release.StageResult(
+        "unit", [], 0, 0.1, "1 skipped", ["SKIPPED [1] x.py:1: exiftool not installed"]
+    )
+    silent = release.StageResult(
+        "unit", [], 0, 0.1, "1 skipped", ["SKIPPED [1] x.py:1: flaky, look later"]
+    )
+
+    assert release.unexpected_skips([environment]) == []
+    assert release.unexpected_skips([silent, environment]) == [
+        "SKIPPED [1] x.py:1: flaky, look later"
+    ]
+
+
+# ── the real-library dry run ─────────────────────────────────────────────────
+
+
+def _library(root: Path) -> None:
+    (root / "album").mkdir(parents=True)
+    (root / "album" / "a.jpg").write_bytes(b"a" * 128)
+    (root / "album" / "b.png").write_bytes(b"b" * 64)
+    (root / "loose.JPG").write_bytes(b"c" * 32)
+    excluded = root / "_IGNORE" / "private"
+    excluded.mkdir(parents=True)
+    (excluded / "secret.jpg").write_bytes(b"never read")
+
+
+def test_the_dry_run_describes_the_library_without_touching_it(tmp_path):
+    config = _config(tmp_path)
+    root = Path(config.library_roots[0])
+    _library(root)
+    before = {
+        str(p): (p.stat().st_mtime_ns, p.read_bytes()) for p in root.rglob("*") if p.is_file()
+    }
+
+    report = release.dry_run(config)
+
+    assert report["files"] == 3, "the excluded sentinel is not counted"
+    assert report["by_extension"] == {".jpg": 2, ".png": 1}
+    assert report["excluded_directories"] >= 1
+    assert report["mutation"] == "none — this pass is read-only"
+    assert report["checksum"]
+    assert not any("secret" in json.dumps(report) for _ in [0]), "excluded content never appears"
+    after = {
+        str(p): (p.stat().st_mtime_ns, p.read_bytes()) for p in root.rglob("*") if p.is_file()
+    }
+    assert after == before, "a read-only pass changed the library"
+
+
+def test_the_dry_run_reconciles_against_what_the_database_already_knows(tmp_path):
+    config = _config(tmp_path)
+    root = Path(config.library_roots[0])
+    _library(root)
+    run_migrations(config.database_url)
+    engine = create_db_engine(config.database_url)
+    with create_session_factory(engine)() as session:
+        session.add(
+            Asset(
+                id=str(uuid.uuid4()),
+                original_path=str(root / "album" / "a.jpg"),
+                current_path=str(root / "album" / "a.jpg"),
+                discovered_at=NOW,
+                hash_version=1,
+                byte_size=128,
+            )
+        )
+        session.add(
+            Asset(
+                id=str(uuid.uuid4()),
+                original_path=str(root / "album" / "gone.jpg"),
+                current_path=str(root / "album" / "gone.jpg"),
+                discovered_at=NOW,
+                hash_version=1,
+                byte_size=1,
+            )
+        )
+        session.commit()
+    engine.dispose()
+
+    reconciliation = release.dry_run(config)["reconciliation"]
+
+    assert reconciliation["known_to_database"] == 2
+    assert reconciliation["already_registered"] == 1
+    assert reconciliation["new_to_the_application"] == 2
+    assert reconciliation["recorded_but_absent_total"] == 1
+    assert reconciliation["recorded_but_absent"][0].endswith("gone.jpg")
+
+
+def test_a_library_root_that_is_not_there_is_refused(tmp_path):
+    config = _config(tmp_path, PHOTO_PIPELINE_LIBRARY_ROOTS=str(tmp_path / "nowhere"))
+    with pytest.raises(release.ReleaseError, match="not a directory"):
+        release.dry_run(config)
+
+
+# ── the approval ─────────────────────────────────────────────────────────────
+
+
+def test_mutation_is_refused_until_the_dry_run_is_approved(tmp_path):
+    config = _config(tmp_path, PHOTO_PIPELINE_REQUIRE_DRY_RUN_APPROVAL="1")
+    _library(Path(config.library_roots[0]))
+
+    blockers = release.mutation_blockers(config)
+    assert [blocker["code"] for blocker in blockers] == ["dry_run_not_approved"]
+
+    record = release.approve(config, release.dry_run(config), approver="domverse")
+    assert record["approved_by"] == "domverse" and record["report_checksum"]
+    assert release.mutation_blockers(config) == []
+
+
+def test_an_approval_covers_the_library_it_was_written_for(tmp_path):
+    config = _config(tmp_path, PHOTO_PIPELINE_REQUIRE_DRY_RUN_APPROVAL="1")
+    _library(Path(config.library_roots[0]))
+    release.approve(config, release.dry_run(config), approver="domverse")
+
+    other = tmp_path / "other-library"
+    other.mkdir()
+    moved = config.model_copy(update={"library_roots": (other,)})
+
+    assert [b["code"] for b in release.mutation_blockers(moved)] == ["approval_scope_mismatch"]
+
+
+def test_without_the_requirement_nothing_changes(tmp_path):
+    config = _config(tmp_path)  # the loopback development default
+    assert release.mutation_blockers(config) == []
+
+
+def test_the_api_refuses_every_mutation_until_the_report_is_approved(tmp_path):
+    config = _config(tmp_path, PHOTO_PIPELINE_REQUIRE_DRY_RUN_APPROVAL="1")
+    _library(Path(config.library_roots[0]))
+    with TestClient(create_app(config)) as client:
+        # Reading stays open: an operator has to see what was found to approve it.
+        assert client.get("/api/v1/workflow").status_code == 200
+        refused = client.post("/api/v1/inventory/scan", json={})
+        assert refused.status_code == 403
+        assert refused.json()["error"]["code"] == "dry_run_not_approved"
+        # A backup is the one mutation a careful operator takes first.
+        assert client.post("/api/v1/backups", json={}).status_code == 201
+
+        release.approve(config, release.dry_run(config), approver="domverse")
+        assert client.post("/api/v1/inventory/scan", json={}).status_code in (200, 201, 202)
+
+
+def test_approving_something_that_is_not_a_report_is_refused(tmp_path):
+    config = _config(tmp_path)
+    with pytest.raises(release.ReleaseError, match="not a dry-run report"):
+        release.approve(config, {"files": 3}, approver="domverse")
+
+
+def test_the_cli_runs_the_dry_run_and_the_approval(tmp_path):
+    config = _config(tmp_path, PHOTO_PIPELINE_REQUIRE_DRY_RUN_APPROVAL="1")
+    _library(Path(config.library_roots[0]))
+    from photo_pipeline.__main__ import main
+
+    environment = {
+        "PHOTO_PIPELINE_DATA_DIR": str(config.data_dir),
+        "PHOTO_PIPELINE_LIBRARY_ROOTS": str(config.library_roots[0]),
+        "PHOTO_PIPELINE_REQUIRE_DRY_RUN_APPROVAL": "1",
+    }
+    previous = {key: os.environ.get(key) for key in environment}
+    os.environ.update(environment)
+    try:
+        report_path = tmp_path / "dry-run.json"
+        assert main(["dry-run", "--output", str(report_path)]) == 0
+        assert json.loads(report_path.read_text())["files"] == 3
+        assert main(["approve-dry-run", str(report_path), "--approver", "domverse"]) == 0
+    finally:
+        for key, value in previous.items():
+            if value is None:
+                os.environ.pop(key, None)
+            else:
+                os.environ[key] = value
+    assert release.mutation_blockers(config) == []

tests/e2e/test_release_journey.py

@@ -0,0 +1,348 @@
+"""The release journey (US07-07): one library, one fresh environment, every stage.
+
+This is the acceptance the whole backlog builds up to — discovery, duplicate review,
+safety, analysis, EXIF verification, album proposal, guarded rename, rescan and
+reconciliation, upload, archive, offline deduplication, restore — driven over HTTP
+against real ``photo_pipeline serve`` and worker child processes, with full process
+restarts in the middle and at the end.
+
+Nothing is reached into. External services are the deterministic fakes the earlier
+phases already use, invoked through the real integration layer: a vision fake that
+records every path it was given, a real fake ``immich-go`` executable, and an
+archive medium that is an ordinary directory whose marker file is its identity.
+
+The invariants asserted along the way are the ones the concept calls non-negotiable:
+
+- an asset's identity survives a rename, an upload, an archive, and a restore;
+- an ``_IGNORE`` sentinel is never discovered, counted, analysed, or uploaded;
+- an NSFW asset never reaches the vision provider but still reaches Immich;
+- no photo's bytes are lost at any point — every hash is still reachable somewhere;
+- every stage's durable state survives a restart of both processes.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import shutil
+from pathlib import Path
+
+import httpx
+import pytest
+
+from tests.e2e._pipeline_harness import (
+    SENTINEL_KEY,
+    FakeImmich,
+    Server,
+    fake_uploader,
+    image,
+    seed_library,
+    start_worker,
+    wait_until,
+)
+
+TIMEOUT = 30
+ALBUM = "rome"
+UPLOADER = 'echo "INFO  uploaded $6"\necho "Uploaded 2, duplicates 0"\nexit 0\n'
+
+
+def _sha256(path: Path) -> str:
+    return hashlib.sha256(path.read_bytes()).hexdigest()
+
+
+def _hashes(*roots: Path) -> set[str]:
+    return {
+        _sha256(path)
+        for root in roots
+        for path in root.rglob("*.jpg")
+        if path.is_file() and not path.name.startswith(".")
+    }
+
+
+def _post(base: str, path: str, **kwargs) -> httpx.Response:
+    response = httpx.post(f"{base}/api/v1{path}", timeout=TIMEOUT, **kwargs)
+    response.raise_for_status()
+    return response
+
+
+def _get(base: str, path: str, **kwargs) -> dict:
+    response = httpx.get(f"{base}/api/v1{path}", timeout=TIMEOUT, **kwargs)
+    response.raise_for_status()
+    return response.json()
+
+
+def _await_job(base: str, job_id: str, *, states=("succeeded",)) -> dict:
+    return wait_until(
+        lambda: (
+            snapshot
+            if (snapshot := _get(base, f"/jobs/{job_id}"))["state"] in states
+            else None
+        ),
+        timeout=90,
+    )
+
+
+@pytest.fixture
+def library(tmp_path):
+    """A fresh library: an album, an exact duplicate, and an excluded sentinel."""
+    seeded = seed_library(tmp_path, {}, {})
+    album = seeded.lib / ALBUM
+    image(album / "a.jpg", 11)
+    image(album / "b.jpg", 12)
+    shutil.copyfile(album / "a.jpg", album / "a-copy.jpg")  # exact duplicate
+    ignored = seeded.lib / "_IGNORE" / "private"
+    ignored.mkdir(parents=True)
+    image(ignored / "sentinel-9f3a2b.jpg", 99)
+    return seeded
+
+
+@pytest.mark.skipif(shutil.which("exiftool") is None, reason="exiftool not installed")
+def test_the_full_release_journey_survives_every_stage_and_two_restarts(library, tmp_path):
+    immich = FakeImmich()
+    uploader = fake_uploader(tmp_path, UPLOADER)
+    vision_log = tmp_path / "vision.log"
+    archive_root = tmp_path / "medium"
+    archive_root.mkdir()
+    environment = {
+        "PHOTO_PIPELINE_IMMICH_SERVER_URL": immich.url,
+        "PHOTO_PIPELINE_IMMICH_API_KEY": SENTINEL_KEY,
+        "PHOTO_PIPELINE_IMMICH_GO_BINARY": str(uploader),
+        "PHOTO_PIPELINE_ARCHIVE_FREE_SPACE_RESERVE_BYTES": "0",
+        "PHOTO_PIPELINE_FAKE_VISION_LOG": str(vision_log),
+    }
+    server = Server(library, extra_env=environment).start()
+    worker = start_worker(library, extra_env=environment)
+    base = server.base
+
+    try:
+        # ── 0. discovery ─────────────────────────────────────────────────────
+        _post(base, "/inventory/scan")
+        assets = _get(base, "/inventory/assets", params={"limit": 200})["items"]
+        assert len(assets) == 3, "the sentinel under _IGNORE is not an asset"
+        paths = {asset["current_path"] for asset in assets}
+        assert not any("_IGNORE" in path or "sentinel" in path for path in paths)
+        identity = {asset["id"]: Path(asset["current_path"]).name for asset in assets}
+
+        # ── 1. duplicate review ──────────────────────────────────────────────
+        _post(base, "/duplicates/detect")
+        clusters = _get(base, "/duplicates/clusters")["items"]
+        assert len(clusters) == 1 and clusters[0]["member_total"] == 2
+        cluster = _get(base, f"/duplicates/clusters/{clusters[0]['id']}")
+        canonical = sorted(member["asset_id"] for member in cluster["members"])[0]
+        _post(
+            base,
+            f"/duplicates/clusters/{cluster['id']}/decision",
+            json={
+                "decision": "canonical",
+                "canonical_asset_id": canonical,
+                "expected_version": cluster["version"],
+            },
+        )
+
+        # ── 2. safety, with its EXIF checkpoint ──────────────────────────────
+        queue = _get(base, "/safety/queue", params={"limit": 100})["items"]
+        assert len(queue) == 2, "a non-canonical variant is not reviewed twice"
+        decisions = {}
+        for index, item in enumerate(sorted(queue, key=lambda row: row["current_path"])):
+            decision = "nsfw" if index == 0 else "sfw"
+            decisions[item["asset_id"]] = decision
+            result = _post(
+                base, "/safety/decisions", json={"asset_id": item["asset_id"], "decision": decision}
+            ).json()
+            assert result["exif_verified"] is True, "the safety checkpoint must verify"
+
+        # ── restart: everything so far has to be durable ─────────────────────
+        server.stop()
+        server.start()
+        base = server.base
+        assert _get(base, "/safety/counts")["nsfw"] == 1
+        assert {a["id"] for a in _get(base, "/inventory/assets", params={"limit": 200})["items"]} == set(
+            identity
+        )
+
+        # ── 3. analysis, gated to confirmed-SFW assets ───────────────────────
+        job = _post(base, "/analysis/jobs").json()
+        _await_job(base, job["id"])
+        analysed = [
+            name
+            for name, decision in (
+                (identity[asset_id], decision) for asset_id, decision in decisions.items()
+            )
+            if decision == "sfw"
+        ]
+        seen = vision_log.read_text().splitlines()
+        assert len(seen) == len(analysed) == 1
+        assert not any("sentinel" in line or "_IGNORE" in line for line in seen)
+        nsfw_id = next(aid for aid, decision in decisions.items() if decision == "nsfw")
+        assert all(identity[nsfw_id] not in line for line in seen), "NSFW reached the provider"
+
+        # From here on no stage may change a photo's bytes: the metadata stages are
+        # done, and moving, uploading, archiving, and restoring only relocate them.
+        stable_hashes = _hashes(library.lib)
+
+        # ── 4. album proposal and guarded rename ─────────────────────────────
+        _post(base, "/albums/proposals", json={})
+        proposal = _get(base, f"/albums/proposals/{ALBUM}")
+        _post(
+            base,
+            f"/albums/proposals/{ALBUM}/edit",
+            json={"name": "2019 Rome", "expected_version": proposal["version"]},
+        )
+        proposal = _get(base, f"/albums/proposals/{ALBUM}")
+        _post(
+            base,
+            f"/albums/proposals/{ALBUM}/approve",
+            json={"expected_version": proposal["version"]},
+        )
+        plan = _post(base, "/rename-plans").json()
+        assert plan["blockers"] == [], [
+            (issue["code"], issue["message"])
+            for op in plan["operations"]
+            for issue in op["issues"]
+        ]
+        response = httpx.post(
+            f"{base}/api/v1/rename-plans/{plan['id']}/apply",
+            json={"expected_version": plan["version"], "expected_checksum": plan["checksum"]},
+            timeout=TIMEOUT,
+        )
+        assert response.status_code == 200, response.text
+        applied = response.json()
+        assert applied["failed"] == 0 and applied["applied"] == 1
+        assert (library.lib / "2019 Rome").is_dir() and not (library.lib / ALBUM).exists()
+
+        # ── 5. rescan and reconciliation: identity survives the move ─────────
+        _post(base, "/inventory/scan")
+        after_rename = _get(base, "/inventory/assets", params={"limit": 200})["items"]
+        assert {asset["id"] for asset in after_rename} == set(identity)
+        assert all("2019 Rome" in asset["current_path"] for asset in after_rename)
+        assert _hashes(library.lib) == stable_hashes, "a rename changed a photo's bytes"
+
+        # ── 6. upload ────────────────────────────────────────────────────────
+        report = _post(base, "/upload-preflight", json={"albums": ["2019 Rome"]}).json()
+        assert report["state"] == "ready", report["blockers"]
+        batch = _post(
+            base,
+            "/upload-batches",
+            json={"albums": ["2019 Rome"], "token": report["token"]},
+        ).json()["batches"][0]
+        started = _post(base, f"/upload-batches/{batch['id']}/start").json()
+        _await_job(base, started["job"]["id"])
+        uploaded = _get(base, f"/upload-batches/{batch['id']}")
+        assert uploaded["state"] == "succeeded"
+        # The uploader said nothing per file, so the outcome is uncertain until the
+        # server itself is asked whether it holds those exact bytes (US05-04).
+        assert uploaded["outcome_state"] == "requires_verification"
+        verified = _post(base, f"/upload-batches/{batch['id']}/verify").json()
+        assert verified["outcome_state"] == "verified", verified
+        uploaded = _get(base, f"/upload-batches/{batch['id']}")
+        # Reviewed NSFW is uploaded; it simply never reached the analyser.
+        assert {item["asset_id"] for item in uploaded["items"]} >= {nsfw_id}
+
+        # ── 7. archive ───────────────────────────────────────────────────────
+        location = _post(
+            base, "/archive-locations", json={"name": "external", "root": str(archive_root)}
+        ).json()
+        preflight = _post(
+            base, "/archive-preflight", json={"location_id": location["id"]}
+        ).json()
+        assert preflight["state"] == "ready", [
+            (asset["asset_id"], asset["blockers"])
+            for album in preflight["albums"]
+            for asset in album["assets"]
+            if asset["blockers"]
+        ] or preflight
+        archive_plan = _post(
+            base,
+            "/archive-plans",
+            json={"location_id": location["id"], "token": preflight["token"]},
+        ).json()
+        uploaded_ids = {item["asset_id"] for item in uploaded["items"]}
+        _post(base, f"/archive-plans/{archive_plan['id']}/apply")
+        wait_until(
+            lambda: all(
+                asset["availability_state"].startswith("archived")
+                for asset in _get(base, "/inventory/assets", params={"limit": 200})["items"]
+                if asset["id"] in uploaded_ids
+            ),
+            timeout=90,
+        )
+        assert _hashes(library.lib, archive_root) == stable_hashes, "archiving lost bytes"
+
+        # ── 8. offline deduplication ─────────────────────────────────────────
+        (archive_root / ".photo-pipeline-archive.json").rename(
+            archive_root / ".photo-pipeline-archive.json.away"
+        )
+        # A copy of an archived photo turns up in the library under its own name —
+        # the real shape of "I re-imported an old card" — so nothing occupies the
+        # path the archived original would be restored to.
+        returned = library.lib / "2019 Rome" / "rediscovered.jpg"
+        returned.parent.mkdir(parents=True, exist_ok=True)
+        archived_copy = next(archive_root.rglob("*.jpg"))
+        shutil.copyfile(archived_copy, returned)
+        _post(base, "/inventory/scan")
+        _post(base, "/duplicates/detect")
+        offline = _get(base, "/inventory/assets", params={"limit": 200})["items"]
+        archived = [a for a in offline if a["availability_state"].startswith("archived")]
+        assert archived, "an unmounted medium must not make assets missing"
+        assert all(a["availability_state"] != "missing_unexpected" for a in offline)
+        assert any(
+            cluster["member_total"] >= 2 for cluster in _get(base, "/duplicates/clusters")["items"]
+        ), "the rediscovered copy did not meet its archived original"
+
+        # ── 9. restore ───────────────────────────────────────────────────────
+        (archive_root / ".photo-pipeline-archive.json.away").rename(
+            archive_root / ".photo-pipeline-archive.json"
+        )
+        restore_report = _post(
+            base, "/restore-preflight", json={"location_id": location["id"]}
+        ).json()
+        restore_plan = _post(
+            base,
+            "/restore-plans",
+            json={"location_id": location["id"], "token": restore_report["token"]},
+        ).json()
+        _post(base, f"/restore-plans/{restore_plan['id']}/apply")
+        wait_until(
+            lambda: all(
+                asset["availability_state"] == "active"
+                for asset in _get(base, "/inventory/assets", params={"limit": 200})["items"]
+                if asset["id"] in identity
+            ),
+            timeout=90,
+        )
+
+        # ── 10. the final restart proves every stage was durable ─────────────
+        worker.kill()
+        worker.wait(timeout=20)
+        server.stop()
+        server.start()
+        base = server.base
+        final = {
+            asset["id"]: asset
+            for asset in _get(base, "/inventory/assets", params={"limit": 200})["items"]
+        }
+        assert set(identity) <= set(final), "an asset id did not survive the journey"
+        assert _get(base, "/safety/counts")["nsfw"] == 1
+        assert _get(base, "/upload-batches")["batches"][0]["state"] == "succeeded"
+        reachable = {
+            _sha256(path): str(path)
+            for root in (library.lib, archive_root)
+            for path in root.rglob("*.jpg")
+            if path.is_file() and not path.name.startswith(".")
+        }
+        assert stable_hashes <= set(reachable), (
+            "a photo was lost",
+            sorted(stable_hashes - set(reachable)),
+            sorted(reachable.values()),
+        )
+        workflow = _get(base, "/workflow")
+        assert {stage["key"] for stage in workflow["stages"]} >= {
+            "inventory",
+            "duplicates",
+            "safety",
+            "analysis",
+        }
+    finally:
+        worker.kill()
+        worker.wait(timeout=20)
+        server.stop()
+        immich.stop()