diff --git a/photo_pipeline/api/app.py b/photo_pipeline/api/app.py
index 742df90..90d711e 100644
--- a/photo_pipeline/api/app.py
+++ b/photo_pipeline/api/app.py
@@ -25,6 +25,7 @@ from photo_pipeline.api.routes import (
     renames,
     safety,
     thumbnails,
+    uploads,
     workflow,
 )
 
@@ -67,6 +68,7 @@ def create_app(config: Config | None = None) -> FastAPI:
     app.include_router(library.router, prefix="/api/v1")
     app.include_router(albums.router, prefix="/api/v1")
     app.include_router(renames.router, prefix="/api/v1")
+    app.include_router(uploads.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")
diff --git a/photo_pipeline/api/routes/uploads.py b/photo_pipeline/api/routes/uploads.py
new file mode 100644
index 0000000..8290295
--- /dev/null
+++ b/photo_pipeline/api/routes/uploads.py
@@ -0,0 +1,39 @@
+"""Upload preflight API (US05-01).
+
+Preflight is a command, not a resource read: it contacts the Immich server, hashes
+the current bytes, and issues a token. It uploads nothing — starting a batch is
+US05-02, so there is deliberately no start endpoint here.
+"""
+
+from __future__ import annotations
+
+from fastapi import APIRouter, Request
+from fastapi.responses import JSONResponse
+from pydantic import BaseModel
+
+from photo_pipeline.services.uploads import UploadError, UploadService
+
+router = APIRouter(tags=["uploads"])
+
+
+class PreflightRequest(BaseModel):
+    # ``None`` means every album; an explicit list scopes the check.
+    albums: list[str] | None = None
+    # Approving a partial upload is an explicit act, never a default.
+    allow_partial: bool = False
+
+
+def _service(request: Request) -> UploadService:
+    return UploadService(request.app.state.session_factory, config=request.app.state.config)
+
+
+@router.post("/upload-preflight")
+def preflight(request: Request, body: PreflightRequest | None = None):
+    body = body or PreflightRequest()
+    try:
+        return _service(request).preflight(body.albums, allow_partial=body.allow_partial)
+    except UploadError as error:
+        return JSONResponse(
+            status_code=422,
+            content={"error": {"code": "unknown_album", "message": str(error)}},
+        )
diff --git a/photo_pipeline/config.py b/photo_pipeline/config.py
index 0bff2eb..3df0980 100644
--- a/photo_pipeline/config.py
+++ b/photo_pipeline/config.py
@@ -37,6 +37,8 @@ class Config(BaseModel):
 
     vision_api_key: SecretStr | None = None
     immich_api_key: SecretStr | None = None
+    immich_server_url: str = ""
+    immich_go_binary: str = "immich-go"
 
     @property
     def database_path(self) -> Path:
diff --git a/photo_pipeline/integrations/immich_go.py b/photo_pipeline/integrations/immich_go.py
new file mode 100644
index 0000000..39081e8
--- /dev/null
+++ b/photo_pipeline/integrations/immich_go.py
@@ -0,0 +1,107 @@
+"""immich-go adapter: binary discovery, version, and redacted command preview.
+
+Upload is the only stage that needs credentials, so this module is also the single
+place that knows an API key exists. It never returns, logs, or renders the secret:
+:func:`build_command` produces the real argument list for the uploader, and
+:func:`redact` produces the copy that is safe for the API, the browser, and the
+activity log. Both come from the same builder so the preview can never drift from
+the command that would actually run.
+
+Server reachability uses ``/api/server/ping`` through stdlib ``urllib`` — the app
+has no HTTP client dependency and this is one request.
+"""
+
+from __future__ import annotations
+
+import json
+import shutil
+import subprocess
+import urllib.error
+import urllib.request
+from pathlib import Path
+
+REDACTED = "***"
+PING_PATH = "/api/server/ping"
+PING_TIMEOUT_SECONDS = 5.0
+
+
+def find_binary(binary: str = "immich-go") -> str | None:
+    """Absolute path of the uploader, or ``None`` when it is not installed."""
+    return shutil.which(binary)
+
+
+def version(binary: str = "immich-go") -> str | None:
+    """Reported uploader version, or ``None`` when it is missing or unusable.
+
+    The version is persisted with every batch (concept §8) because immich-go's
+    flags and report text change between releases.
+    """
+    path = find_binary(binary)
+    if path is None:
+        return None
+    try:
+        result = subprocess.run([path, "--version"], capture_output=True, text=True, timeout=30)
+    except (OSError, subprocess.SubprocessError):
+        return None
+    output = (result.stdout or result.stderr or "").strip()
+    return output.splitlines()[0].strip() if output else None
+
+
+def ping(server_url: str, *, timeout: float = PING_TIMEOUT_SECONDS) -> tuple[bool, str | None]:
+    """``(reachable, detail)`` for the configured Immich server.
+
+    A reachable Immich answers ``{"res": "pong"}``. Anything else — wrong host, no
+    Immich, HTTP error — is a blocker with a short human detail. The detail never
+    carries the URL's credentials because the key travels in a header, not the URL.
+    """
+    if not server_url:
+        return False, "no server URL configured"
+    url = server_url.rstrip("/") + PING_PATH
+    try:
+        with urllib.request.urlopen(url, timeout=timeout) as response:  # noqa: S310
+            payload = json.loads(response.read().decode("utf-8") or "{}")
+    except (urllib.error.URLError, OSError, ValueError, TimeoutError) as error:
+        return False, f"{type(error).__name__}: {error}"
+    if payload.get("res") == "pong":
+        return True, None
+    return False, "server did not answer with pong"
+
+
+def build_command(
+    *,
+    binary: str,
+    server_url: str,
+    api_key: str,
+    album_name: str,
+    folder: Path | str,
+) -> list[str]:
+    """The exact upload invocation for one folder-as-album batch."""
+    return [
+        binary,
+        "upload",
+        "from-folder",
+        f"--server={server_url}",
+        f"--api-key={api_key}",
+        f"--album-name={album_name}",
+        str(folder),
+    ]
+
+
+def redact(command: list[str]) -> list[str]:
+    """The same command with every secret-bearing argument masked."""
+    return [f"--api-key={REDACTED}" if arg.startswith("--api-key=") else arg for arg in command]
+
+
+def preview_command(
+    *, binary: str, server_url: str, album_name: str, folder: Path | str
+) -> list[str]:
+    """Redacted preview built without ever handling the real key."""
+    return redact(
+        build_command(
+            binary=binary,
+            server_url=server_url,
+            api_key=REDACTED,
+            album_name=album_name,
+            folder=folder,
+        )
+    )
diff --git a/photo_pipeline/services/uploads.py b/photo_pipeline/services/uploads.py
new file mode 100644
index 0000000..bea6090
--- /dev/null
+++ b/photo_pipeline/services/uploads.py
@@ -0,0 +1,306 @@
+"""UploadService — preflight for Immich upload (US05-01).
+
+Upload is the first stage that sends the library somewhere the app cannot take it
+back from, so nothing here uploads: preflight only *proves* a scope is safe and
+issues a token that a later start command must present (US05-02).
+
+What it proves (concept §8 "preflight"):
+
+- credentials are configured and the Immich server answers, without ever putting
+  the API key in a response, a preview, or a log line;
+- ``immich-go`` is installed and its version is recorded;
+- no rename is half-applied — the journal must not block library mutation;
+- every asset in scope is canonical, present, and decided ``sfw``/``nsfw`` with a
+  verified safety EXIF checkpoint;
+- every SFW asset also has a completed analysis EXIF checkpoint;
+- the bytes on disk right now still hash to what inventory recorded, so the album
+  preview describes the exact bytes that would be uploaded.
+
+Blocker codes are structured, never prose the UI has to parse:
+
+``no_library_root``, ``credentials_missing``, ``server_unreachable``,
+``immich_go_missing``, ``rename_pending``, ``unknown_album``, ``empty_scope``,
+``file_missing``, ``bytes_changed``, ``safety_undecided``, ``safety_deferred``,
+``safety_exif_unverified``, ``analysis_incomplete``, ``partial_scope``.
+
+**Partial scope is a blocker, not a default.** An album with any blocked asset is
+refused unless the caller passes the explicit ``allow_partial`` policy, which is
+itself part of the token — a token issued for a partial upload can never be
+replayed as a full one.
+
+The token is derived, not stored: it is a digest of the whole report (minus its
+timestamp), so any change that matters — a new decision, edited bytes, a different
+server, a resolved blocker, a different policy — produces a different token and the
+old one is stale by construction. No table, no invalidation bookkeeping.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+from datetime import datetime, timezone
+from pathlib import Path
+
+from sqlalchemy import select
+from sqlalchemy.orm import sessionmaker
+
+from photo_pipeline.config import Config
+from photo_pipeline.integrations import immich_go
+from photo_pipeline.models import AnalysisResult, Asset, SafetyReview
+from photo_pipeline.services.albums import album_label
+from photo_pipeline.services.hashing import sha256_file
+from photo_pipeline.services.rename_journal import RenameJournal
+
+PREFLIGHT_VERSION = 1
+TOKEN_PREFIX = f"v{PREFLIGHT_VERSION}"
+SFW = "sfw"
+NSFW = "nsfw"
+ANALYZED = "analyzed"
+
+
+class UploadError(RuntimeError):
+    """Invalid preflight request (unknown album in the requested scope)."""
+
+
+def _now() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+def _issue(code: str, message: str) -> dict:
+    return {"code": code, "message": message}
+
+
+class UploadService:
+    def __init__(self, session_factory: sessionmaker, *, config: Config) -> None:
+        self._session_factory = session_factory
+        self._config = config
+        self._roots = tuple(Path(root) for root in config.library_roots)
+
+    # ── preflight ─────────────────────────────────────────────────────────────
+
+    def preflight(self, albums: list[str] | None = None, *, allow_partial: bool = False) -> dict:
+        """Validate an upload scope and issue its token. Read-only, no upload."""
+        report = {
+            "schema_version": PREFLIGHT_VERSION,
+            "policy": {"allow_partial": allow_partial},
+            "blockers": [],
+            "credentials": self._credentials(),
+            "server": self._server(),
+            "uploader": self._uploader(),
+        }
+        report["blockers"] += self._environment_blockers(report)
+        report["albums"] = self._albums(albums, allow_partial=allow_partial)
+        report["totals"] = _totals(report["albums"])
+        if not report["albums"]:
+            report["blockers"].append(
+                _issue("empty_scope", "no canonical, active assets are in the selected scope")
+            )
+        report["state"] = (
+            "ready"
+            if not report["blockers"] and all(a["state"] == "ready" for a in report["albums"])
+            else "blocked"
+        )
+        report["token"] = _token(report)
+        report["generated_at"] = _now().isoformat()
+        return report
+
+    def verify_token(
+        self, token: str, albums: list[str] | None = None, *, allow_partial: bool = False
+    ) -> bool:
+        """True when ``token`` still describes the current state of that scope.
+
+        Recomputed rather than looked up, so an externally edited file or a changed
+        decision invalidates it even though nothing wrote to the database.
+        """
+        return bool(token) and token == self.preflight(albums, allow_partial=allow_partial)["token"]
+
+    # ── environment ───────────────────────────────────────────────────────────
+
+    def _credentials(self) -> dict:
+        """Presence only — the key itself never leaves configuration."""
+        return {
+            "server_url": self._config.immich_server_url,
+            "api_key_configured": self._config.immich_api_key is not None,
+        }
+
+    def _server(self) -> dict:
+        reachable, detail = immich_go.ping(self._config.immich_server_url)
+        return {"reachable": reachable, "detail": detail}
+
+    def _uploader(self) -> dict:
+        binary = self._config.immich_go_binary
+        return {
+            "binary": binary,
+            "installed": immich_go.find_binary(binary) is not None,
+            "version": immich_go.version(binary),
+        }
+
+    def _environment_blockers(self, report: dict) -> list[dict]:
+        blockers: list[dict] = []
+        if not self._roots:
+            blockers.append(_issue("no_library_root", "no library root is configured"))
+        if not report["credentials"]["api_key_configured"] or not self._config.immich_server_url:
+            blockers.append(
+                _issue("credentials_missing", "an Immich server URL and API key are required")
+            )
+        elif not report["server"]["reachable"]:
+            blockers.append(
+                _issue(
+                    "server_unreachable", f"Immich did not respond: {report['server']['detail']}"
+                )
+            )
+        if not report["uploader"]["installed"]:
+            blockers.append(
+                _issue("immich_go_missing", f"{self._config.immich_go_binary} is not installed")
+            )
+        if RenameJournal(self._session_factory).blocks_mutation():
+            blockers.append(
+                _issue("rename_pending", "an unresolved rename must be recovered before upload")
+            )
+        return blockers
+
+    # ── scope ─────────────────────────────────────────────────────────────────
+
+    def _albums(self, requested: list[str] | None, *, allow_partial: bool) -> list[dict]:
+        by_album = self._scope()
+        if requested is not None:
+            unknown = sorted(set(requested) - set(by_album))
+            if unknown:
+                raise UploadError(f"unknown album(s): {', '.join(unknown)}")
+            by_album = {name: by_album[name] for name in sorted(set(requested))}
+        return [
+            self._album(name, rows, allow_partial=allow_partial)
+            for name, rows in sorted(by_album.items())
+        ]
+
+    def _scope(self) -> dict[str, list[dict]]:
+        """Canonical, active assets grouped by album, each with its stage evidence."""
+        with self._session_factory() as session:
+            assets = list(
+                session.scalars(
+                    select(Asset).where(
+                        Asset.canonical_asset_id.is_(None),
+                        Asset.availability_state == "active",
+                        Asset.current_path.is_not(None),
+                    )
+                )
+            )
+            reviews: dict[str, SafetyReview] = {}
+            for review in session.scalars(select(SafetyReview).order_by(SafetyReview.created_at)):
+                reviews[review.asset_id] = review  # latest row per asset wins
+            analyses = {
+                result.asset_id: result for result in session.scalars(select(AnalysisResult))
+            }
+
+        by_album: dict[str, list[dict]] = {}
+        for asset in assets:
+            by_album.setdefault(album_label(asset.current_path, self._roots), []).append(
+                {
+                    "asset_id": asset.id,
+                    "path": asset.current_path,
+                    "expected_sha256": asset.current_sha256,
+                    "review": reviews.get(asset.id),
+                    "analysis": analyses.get(asset.id),
+                }
+            )
+        return by_album
+
+    def _album(self, name: str, rows: list[dict], *, allow_partial: bool) -> dict:
+        folder = Path(rows[0]["path"]).parent
+        items = sorted((self._item(row) for row in rows), key=lambda item: item["current_path"])
+        blocked = [item for item in items if item["blockers"]]
+        eligible = [item for item in items if not item["blockers"]]
+        blockers: list[dict] = []
+        if blocked and not allow_partial:
+            blockers.append(
+                _issue(
+                    "partial_scope",
+                    f"{len(blocked)} of {len(items)} asset(s) are not upload-ready; resolve them "
+                    "or approve a partial upload explicitly",
+                )
+            )
+        if not eligible:
+            blockers.append(_issue("empty_scope", "no upload-ready asset remains in this album"))
+        # Folder-as-album: Immich names the album after the leaf folder, so the
+        # preview shows exactly what the server will create.
+        album_name = folder.name
+        return {
+            "album": name,
+            "folder": str(folder),
+            "album_name": album_name,
+            "asset_count": len(items),
+            "eligible_count": len(eligible),
+            "blocked_count": len(blocked),
+            "partial": bool(blocked),
+            "state": "blocked" if blockers else "ready",
+            "blockers": blockers,
+            "assets": items,
+            "command_preview": immich_go.preview_command(
+                binary=self._config.immich_go_binary,
+                server_url=self._config.immich_server_url,
+                album_name=album_name,
+                folder=folder,
+            ),
+        }
+
+    def _item(self, row: dict) -> dict:
+        """One asset's readiness, including its hash *as it is on disk right now*."""
+        path = Path(row["path"])
+        review: SafetyReview | None = row["review"]
+        analysis: AnalysisResult | None = row["analysis"]
+        decision = review.decision if review else None
+        blockers: list[dict] = []
+        current_sha256 = None
+
+        if not path.exists():
+            blockers.append(_issue("file_missing", f"{path} is missing"))
+        else:
+            # ponytail: full re-hash every preflight. Gate on (size, mtime_ns) first
+            # if a large library makes this slow — the hash stays authoritative.
+            current_sha256 = sha256_file(path)
+            if row["expected_sha256"] and current_sha256 != row["expected_sha256"]:
+                blockers.append(
+                    _issue("bytes_changed", f"{path} changed since its last verified checkpoint")
+                )
+
+        if decision not in (SFW, NSFW):
+            code = "safety_deferred" if decision == "deferred" else "safety_undecided"
+            blockers.append(_issue(code, "a confirmed sfw/nsfw safety decision is required"))
+        elif review.exif_verified_at is None:
+            blockers.append(
+                _issue("safety_exif_unverified", "the safety EXIF checkpoint is not verified")
+            )
+        elif decision == SFW and not (
+            analysis and analysis.status == ANALYZED and analysis.exif_written_at is not None
+        ):
+            blockers.append(
+                _issue("analysis_incomplete", "SFW assets need a verified analysis EXIF checkpoint")
+            )
+
+        return {
+            "asset_id": row["asset_id"],
+            "current_path": str(path),
+            "safety_decision": decision,
+            "current_sha256": current_sha256,
+            "blockers": blockers,
+        }
+
+
+def _totals(albums: list[dict]) -> dict:
+    return {
+        "albums": len(albums),
+        "ready_albums": sum(1 for album in albums if album["state"] == "ready"),
+        "assets": sum(album["asset_count"] for album in albums),
+        "eligible": sum(album["eligible_count"] for album in albums),
+        "blocked": sum(album["blocked_count"] for album in albums),
+    }
+
+
+def _token(report: dict) -> str:
+    """Digest of everything the report asserts. Volatile fields are excluded so the
+    same state always yields the same token; every relevant change breaks it."""
+    payload = {key: value for key, value in report.items() if key not in ("generated_at", "token")}
+    digest = hashlib.sha256(
+        json.dumps(payload, sort_keys=True, ensure_ascii=False, default=str).encode("utf-8")
+    ).hexdigest()
+    return f"{TOKEN_PREFIX}:{digest}"
diff --git a/tests/integration/test_upload_preflight.py b/tests/integration/test_upload_preflight.py
new file mode 100644
index 0000000..28349e0
--- /dev/null
+++ b/tests/integration/test_upload_preflight.py
@@ -0,0 +1,511 @@
+"""Upload preflight: credentials, scope, readiness, and tokens (US05-01).
+
+Every external boundary is faked but never mocked away: the uploader is a real
+executable on disk invoked through ``subprocess``, and the Immich server is a real
+localhost HTTP server answering ``/api/server/ping``. Preflight itself must stay
+read-only — the library snapshot is asserted unchanged.
+"""
+
+import json
+import stat
+import threading
+import uuid
+from datetime import datetime, timezone
+from http.server import BaseHTTPRequestHandler, HTTPServer
+
+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 (
+    AnalysisResult,
+    Asset,
+    RenameOperation,
+    RenamePlan,
+    SafetyReview,
+)
+from photo_pipeline.services.hashing import sha256_file
+from photo_pipeline.services.uploads import UploadError, UploadService
+
+NOW = datetime(2026, 1, 1, tzinfo=timezone.utc)
+# A sentinel credential: every assertion below proves it never leaves configuration.
+SENTINEL_KEY = "immich-sentinel-9f3a2b"
+UPLOADER_VERSION = "immich-go 0.21.0"
+
+
+# ── fake external boundary ───────────────────────────────────────────────────
+
+
+class _PingHandler(BaseHTTPRequestHandler):
+    payload = b'{"res":"pong"}'
+    status = 200
+
+    def do_GET(self):  # noqa: N802 (BaseHTTPRequestHandler API)
+        self.send_response(type(self).status)
+        self.send_header("Content-Type", "application/json")
+        self.end_headers()
+        self.wfile.write(type(self).payload)
+
+    def log_message(self, *args):
+        pass  # keep the test output clean
+
+
+@pytest.fixture
+def immich_server():
+    """A real HTTP server that answers like Immich. Yields its base URL."""
+    handler = type("Handler", (_PingHandler,), {})
+    server = HTTPServer(("127.0.0.1", 0), handler)
+    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    yield f"http://127.0.0.1:{server.server_port}", handler
+    server.shutdown()
+    server.server_close()
+
+
+def _fake_uploader(tmp_path):
+    """A real executable standing in for immich-go."""
+    path = tmp_path / "immich-go"
+    path.write_text(f"#!/bin/sh\necho '{UPLOADER_VERSION}'\n")
+    path.chmod(path.stat().st_mode | stat.S_IEXEC | stat.S_IXGRP | stat.S_IXOTH)
+    return path
+
+
+# ── environment ──────────────────────────────────────────────────────────────
+
+
+def _env(tmp_path, server_url, *, credential=SENTINEL_KEY, uploader=None):
+    (tmp_path / "data").mkdir(exist_ok=True)
+    lib = tmp_path / "lib"
+    lib.mkdir(exist_ok=True)
+    env = {
+        "PHOTO_PIPELINE_DATA_DIR": str(tmp_path / "data"),
+        "PHOTO_PIPELINE_LIBRARY_ROOTS": str(lib),
+        "PHOTO_PIPELINE_IMMICH_SERVER_URL": server_url,
+        "PHOTO_PIPELINE_IMMICH_GO_BINARY": str(
+            uploader if uploader is not None else _fake_uploader(tmp_path)
+        ),
+    }
+    if credential:
+        env["PHOTO_PIPELINE_IMMICH_API_KEY"] = credential
+    config = Config.from_env(env)
+    run_migrations(config.database_url)
+    return config, create_session_factory(create_db_engine(config.database_url)), lib
+
+
+def _album(
+    sf,
+    lib,
+    album="rome",
+    names=("a.jpg", "b.jpg"),
+    *,
+    decision="sfw",
+    exif_verified=True,
+    analyzed=True,
+):
+    """A real album folder with registered, stage-complete assets."""
+    folder = lib / album
+    folder.mkdir(parents=True, exist_ok=True)
+    ids = []
+    with sf() as session:
+        for name in names:
+            path = folder / name
+            path.write_bytes(name.encode() * 16)
+            asset_id = str(uuid.uuid4())
+            ids.append(asset_id)
+            session.add(
+                Asset(
+                    id=asset_id,
+                    original_path=str(path),
+                    current_path=str(path),
+                    discovered_at=NOW,
+                    hash_version=1,
+                    byte_size=path.stat().st_size,
+                    current_sha256=sha256_file(path),
+                )
+            )
+            if decision is not None:
+                session.add(
+                    SafetyReview(
+                        id=str(uuid.uuid4()),
+                        asset_id=asset_id,
+                        decision=decision,
+                        exif_verified_at=NOW if exif_verified else None,
+                    )
+                )
+            if analyzed:
+                session.add(
+                    AnalysisResult(
+                        asset_id=asset_id,
+                        status="analyzed",
+                        exif_written_at=NOW,
+                    )
+                )
+        session.commit()
+    return folder, ids
+
+
+def _snapshot(lib):
+    return {
+        str(p.relative_to(lib)): (p.read_bytes() if p.is_file() else None)
+        for p in sorted(lib.rglob("*"))
+    }
+
+
+def _service(sf, config):
+    return UploadService(sf, config=config)
+
+
+def _codes(report):
+    return {issue["code"] for issue in report["blockers"]} | {
+        issue["code"] for album in report["albums"] for issue in album["blockers"]
+    } | {
+        issue["code"]
+        for album in report["albums"]
+        for asset in album["assets"]
+        for issue in asset["blockers"]
+    }
+
+
+# ── happy path ───────────────────────────────────────────────────────────────
+
+
+def test_ready_preflight_reports_scope_command_and_hashes(tmp_path, immich_server):
+    url, _ = immich_server
+    config, sf, lib = _env(tmp_path, url)
+    folder, ids = _album(sf, lib)
+    before = _snapshot(lib)
+
+    report = _service(sf, config).preflight()
+
+    assert report["state"] == "ready" and report["blockers"] == []
+    assert report["server"]["reachable"] is True
+    assert report["uploader"]["installed"] is True
+    assert report["uploader"]["version"] == UPLOADER_VERSION
+    assert report["credentials"]["api_key_configured"] is True
+    assert report["totals"] == {
+        "albums": 1,
+        "ready_albums": 1,
+        "assets": 2,
+        "eligible": 2,
+        "blocked": 0,
+    }
+    album = report["albums"][0]
+    assert album["album"] == "rome" and album["folder"] == str(folder)
+    assert album["album_name"] == "rome"  # folder-as-album
+    assert album["partial"] is False and album["state"] == "ready"
+    assert sorted(a["asset_id"] for a in album["assets"]) == sorted(ids)
+    # Hashes are of the bytes on disk right now, not a remembered value.
+    for asset in album["assets"]:
+        assert asset["current_sha256"] == sha256_file(asset["current_path"])
+    assert report["token"].startswith("v1:")
+    assert _snapshot(lib) == before, "preflight must not touch the library"
+
+
+def test_command_preview_shows_folder_and_album_without_the_key(tmp_path, immich_server):
+    url, _ = immich_server
+    config, sf, lib = _env(tmp_path, url)
+    folder, _ = _album(sf, lib)
+
+    preview = _service(sf, config).preflight()["albums"][0]["command_preview"]
+
+    assert f"--api-key={'***'}" in preview
+    assert "--album-name=rome" in preview
+    assert str(folder) in preview
+    assert SENTINEL_KEY not in " ".join(preview)
+
+
+def test_scoping_to_one_album_excludes_the_others(tmp_path, immich_server):
+    url, _ = immich_server
+    config, sf, lib = _env(tmp_path, url)
+    _album(sf, lib, "rome")
+    _album(sf, lib, "paris", names=("c.jpg",))
+
+    report = _service(sf, config).preflight(["paris"])
+
+    assert [album["album"] for album in report["albums"]] == ["paris"]
+    assert report["totals"]["assets"] == 1
+
+
+def test_unknown_album_in_scope_is_refused(tmp_path, immich_server):
+    url, _ = immich_server
+    config, sf, lib = _env(tmp_path, url)
+    _album(sf, lib, "rome")
+
+    with pytest.raises(UploadError):
+        _service(sf, config).preflight(["atlantis"])
+
+
+def test_empty_scope_is_a_blocker(tmp_path, immich_server):
+    url, _ = immich_server
+    config, sf, _ = _env(tmp_path, url)
+
+    report = _service(sf, config).preflight()
+
+    assert report["state"] == "blocked" and "empty_scope" in _codes(report)
+
+
+# ── credentials and environment ──────────────────────────────────────────────
+
+
+def test_missing_api_key_blocks_without_touching_the_server(tmp_path, immich_server):
+    url, _ = immich_server
+    config, sf, lib = _env(tmp_path, url, credential=None)
+    _album(sf, lib)
+
+    report = _service(sf, config).preflight()
+
+    assert report["state"] == "blocked"
+    assert "credentials_missing" in _codes(report)
+    assert report["credentials"]["api_key_configured"] is False
+
+
+def test_unreachable_server_blocks(tmp_path, immich_server):
+    url, _ = immich_server
+    config, sf, lib = _env(tmp_path, url.replace(url.rsplit(":", 1)[-1], "1"))
+    _album(sf, lib)
+
+    report = _service(sf, config).preflight()
+
+    assert "server_unreachable" in _codes(report)
+    assert report["server"]["reachable"] is False and report["server"]["detail"]
+
+
+def test_server_that_is_not_immich_blocks(tmp_path, immich_server):
+    url, handler = immich_server
+    handler.payload = b'{"error":"unauthorized"}'
+    handler.status = 401
+    config, sf, lib = _env(tmp_path, url)
+    _album(sf, lib)
+
+    assert "server_unreachable" in _codes(_service(sf, config).preflight())
+
+
+def test_missing_uploader_blocks(tmp_path, immich_server):
+    url, _ = immich_server
+    config, sf, lib = _env(tmp_path, url, uploader=tmp_path / "does-not-exist")
+    _album(sf, lib)
+
+    report = _service(sf, config).preflight()
+
+    assert "immich_go_missing" in _codes(report)
+    assert report["uploader"]["installed"] is False and report["uploader"]["version"] is None
+
+
+def test_half_applied_rename_blocks_upload(tmp_path, immich_server):
+    """Album paths must be final before upload: an operation that may have already
+    touched the disk blocks every other mutation until it is recovered (US04-04)."""
+    url, _ = immich_server
+    config, sf, lib = _env(tmp_path, url)
+    folder, _ = _album(sf, lib)
+    with sf() as session:
+        plan_id = str(uuid.uuid4())
+        session.add(RenamePlan(id=plan_id, state="applying", operation_count=1))
+        session.flush()
+        session.add(
+            RenameOperation(
+                id=str(uuid.uuid4()),
+                plan_id=plan_id,
+                sequence=0,
+                operation="move_folder",
+                source_path=str(folder),
+                destination_path=str(lib / "2019 Rome"),
+                journal_state="moving",
+            )
+        )
+        session.commit()
+
+    report = _service(sf, config).preflight()
+
+    assert report["state"] == "blocked" and "rename_pending" in _codes(report)
+
+
+def test_no_secret_appears_anywhere_in_the_report(tmp_path, immich_server):
+    url, _ = immich_server
+    config, sf, lib = _env(tmp_path, url)
+    _album(sf, lib)
+
+    report = _service(sf, config).preflight()
+
+    assert SENTINEL_KEY not in json.dumps(report, default=str)
+
+
+# ── stage readiness ──────────────────────────────────────────────────────────
+
+
+@pytest.mark.parametrize(
+    "kwargs,code",
+    [
+        ({"decision": None}, "safety_undecided"),
+        ({"decision": "deferred"}, "safety_deferred"),
+        ({"exif_verified": False}, "safety_exif_unverified"),
+        ({"analyzed": False}, "analysis_incomplete"),
+    ],
+)
+def test_blocked_stage_blocks_upload(tmp_path, immich_server, kwargs, code):
+    url, _ = immich_server
+    config, sf, lib = _env(tmp_path, url)
+    _album(sf, lib, **kwargs)
+
+    report = _service(sf, config).preflight()
+
+    assert report["state"] == "blocked"
+    assert code in _codes(report)
+    assert report["albums"][0]["eligible_count"] == 0
+
+
+def test_reviewed_nsfw_asset_is_upload_eligible_without_analysis(tmp_path, immich_server):
+    """NSFW never reaches the analyser, but a verified nsfw keyword makes it
+    uploadable (concept §8 eligibility table)."""
+    url, _ = immich_server
+    config, sf, lib = _env(tmp_path, url)
+    _album(sf, lib, decision="nsfw", analyzed=False)
+
+    report = _service(sf, config).preflight()
+
+    assert report["state"] == "ready"
+    assert report["albums"][0]["eligible_count"] == 2
+
+
+def test_changed_bytes_block_the_album(tmp_path, immich_server):
+    url, _ = immich_server
+    config, sf, lib = _env(tmp_path, url)
+    folder, _ = _album(sf, lib)
+    (folder / "a.jpg").write_bytes(b"edited after the checkpoint")
+
+    report = _service(sf, config).preflight()
+
+    assert "bytes_changed" in _codes(report)
+    assert report["albums"][0]["blocked_count"] == 1
+
+
+def test_missing_file_blocks_the_album(tmp_path, immich_server):
+    url, _ = immich_server
+    config, sf, lib = _env(tmp_path, url)
+    folder, _ = _album(sf, lib)
+    (folder / "a.jpg").unlink()
+
+    assert "file_missing" in _codes(_service(sf, config).preflight())
+
+
+# ── partial scope ────────────────────────────────────────────────────────────
+
+
+def test_partial_album_is_blocked_until_explicitly_approved(tmp_path, immich_server):
+    url, _ = immich_server
+    config, sf, lib = _env(tmp_path, url)
+    folder, _ = _album(sf, lib)
+    (folder / "a.jpg").write_bytes(b"changed")
+
+    blocked = _service(sf, config).preflight()
+    approved = _service(sf, config).preflight(allow_partial=True)
+
+    assert blocked["state"] == "blocked"
+    assert "partial_scope" in {issue["code"] for issue in blocked["albums"][0]["blockers"]}
+    assert approved["state"] == "ready"
+    assert approved["albums"][0]["partial"] is True
+    assert approved["albums"][0]["eligible_count"] == 1
+    # The approval is part of the token, so it can never be replayed as a full run.
+    assert approved["token"] != blocked["token"]
+
+
+def test_partial_approval_cannot_rescue_an_album_with_nothing_ready(tmp_path, immich_server):
+    url, _ = immich_server
+    config, sf, lib = _env(tmp_path, url)
+    _album(sf, lib, decision=None)
+
+    report = _service(sf, config).preflight(allow_partial=True)
+
+    assert report["state"] == "blocked"
+    assert "empty_scope" in {issue["code"] for issue in report["albums"][0]["blockers"]}
+
+
+# ── token ────────────────────────────────────────────────────────────────────
+
+
+def test_token_is_stable_while_nothing_relevant_changes(tmp_path, immich_server):
+    url, _ = immich_server
+    config, sf, lib = _env(tmp_path, url)
+    _album(sf, lib)
+    service = _service(sf, config)
+
+    first = service.preflight()["token"]
+
+    assert service.preflight()["token"] == first
+    assert service.verify_token(first) is True
+
+
+@pytest.mark.parametrize("scope", [None, ["rome"]])
+def test_edited_bytes_make_the_token_stale(tmp_path, immich_server, scope):
+    url, _ = immich_server
+    config, sf, lib = _env(tmp_path, url)
+    folder, _ = _album(sf, lib)
+    service = _service(sf, config)
+    token = service.preflight(scope)["token"]
+
+    (folder / "b.jpg").write_bytes(b"edited outside the app")
+
+    assert service.verify_token(token, scope) is False
+
+
+def test_changed_decision_makes_the_token_stale(tmp_path, immich_server):
+    url, _ = immich_server
+    config, sf, lib = _env(tmp_path, url)
+    _, ids = _album(sf, lib)
+    service = _service(sf, config)
+    token = service.preflight()["token"]
+
+    with sf() as session:
+        session.add(
+            SafetyReview(
+                id=str(uuid.uuid4()),
+                asset_id=ids[0],
+                decision="deferred",
+                created_at=datetime(2099, 1, 1, tzinfo=timezone.utc),  # the latest review wins
+            )
+        )
+        session.commit()
+
+    assert service.verify_token(token) is False
+
+
+def test_token_from_a_different_scope_is_rejected(tmp_path, immich_server):
+    url, _ = immich_server
+    config, sf, lib = _env(tmp_path, url)
+    _album(sf, lib, "rome")
+    _album(sf, lib, "paris", names=("c.jpg",))
+    service = _service(sf, config)
+
+    assert service.verify_token(service.preflight(["rome"])["token"], ["paris"]) is False
+    assert service.verify_token("v1:not-a-real-token") is False
+    assert service.verify_token("") is False
+
+
+# ── API surface ──────────────────────────────────────────────────────────────
+
+
+def test_api_preflight_returns_the_report_without_secrets(tmp_path, immich_server):
+    url, _ = immich_server
+    config, sf, lib = _env(tmp_path, url)
+    _album(sf, lib)
+
+    with TestClient(create_app(config)) as client:
+        response = client.post("/api/v1/upload-preflight", json={})
+
+    assert response.status_code == 200
+    body = response.json()
+    assert body["state"] == "ready" and body["token"].startswith("v1:")
+    assert SENTINEL_KEY not in response.text
+
+
+def test_api_rejects_an_unknown_album(tmp_path, immich_server):
+    url, _ = immich_server
+    config, sf, lib = _env(tmp_path, url)
+    _album(sf, lib)
+
+    with TestClient(create_app(config)) as client:
+        response = client.post("/api/v1/upload-preflight", json={"albums": ["atlantis"]})
+
+    assert response.status_code == 422
+    assert response.json()["error"]["code"] == "unknown_album"
diff --git a/tests/story_traceability.json b/tests/story_traceability.json
index 2706256..51312c9 100644
--- a/tests/story_traceability.json
+++ b/tests/story_traceability.json
@@ -103,6 +103,9 @@
     ],
     "US04-06": [
       "tests/e2e/test_phase_d_pipeline.py"
+    ],
+    "US05-01": [
+      "tests/integration/test_upload_preflight.py"
     ]
   }
 }