US04-06: Automate Phase D End-to-End Acceptance

2026-08-16 12:05:32 +02:00
17 changed files with 8 additions and 2410 deletions
--- a/migrations/versions/0008_upload_batches.py
+++ b/migrations/versions/0008_upload_batches.py
@@ -1,91 +0,0 @@
 """Upload batches and their items (US05-02).
 Revision ID: 0008_upload_batches
 Revises: 0007_rename_plans
 Create Date: 2026-08-16
 One row per approved album folder handed to immich-go, plus the per-asset
 pre-upload hashes that later stories verify the result against.
 """
 import sqlalchemy as sa
 from alembic import op
 revision = "0008_upload_batches"
 down_revision = "0007_rename_plans"
 branch_labels = None
 depends_on = None
 def upgrade() -> None:
    op.create_table(
        "upload_batches",
        sa.Column("id", sa.String(), primary_key=True),
        sa.Column("album", sa.String(), nullable=False),
        sa.Column("folder", sa.String(), nullable=False),
        sa.Column("album_name", sa.String(), nullable=False),
        # planned | running | cancelling | cancelled | succeeded | failed
        # | unknown_requires_verification
        sa.Column("state", sa.String(), nullable=False, server_default="planned"),
        # Preflight token this batch was approved against; re-checked before every
        # attempt, so changed bytes or decisions cannot be uploaded silently.
        sa.Column("preflight_token", sa.String(), nullable=False),
        sa.Column("allow_partial", sa.Boolean(), nullable=False, server_default="0"),
        sa.Column("command", sa.String(), nullable=True),  # JSON array, redacted
        sa.Column("uploader_version", sa.String(), nullable=True),
        sa.Column("asset_count", sa.Integer(), nullable=False, server_default="0"),
        sa.Column("attempt_count", sa.Integer(), nullable=False, server_default="0"),
        # Bumped on every claim and used as the fencing token.
        sa.Column("version", sa.Integer(), nullable=False, server_default="1"),
        sa.Column("worker_id", sa.String(), nullable=True),
        sa.Column("report_path", sa.String(), nullable=True),
        sa.Column("report_bytes", sa.Integer(), nullable=True),
        sa.Column("report_truncated", sa.Boolean(), nullable=False, server_default="0"),
        sa.Column("exit_code", sa.Integer(), nullable=True),
        sa.Column("error_code", sa.String(), nullable=True),
        sa.Column("error_message", sa.String(), nullable=True),
        sa.Column("started_at", sa.DateTime(timezone=True), nullable=True),
        sa.Column("finished_at", sa.DateTime(timezone=True), nullable=True),
        sa.Column(
            "created_at",
            sa.DateTime(timezone=True),
            nullable=False,
            server_default=sa.text("CURRENT_TIMESTAMP"),
        ),
        sa.Column(
            "updated_at",
            sa.DateTime(timezone=True),
            nullable=False,
            server_default=sa.text("CURRENT_TIMESTAMP"),
        ),
    )
    op.create_index("ix_upload_batches_album", "upload_batches", ["album"])
    op.create_index("ix_upload_batches_state", "upload_batches", ["state"])
    op.create_table(
        "upload_items",
        sa.Column(
            "batch_id",
            sa.String(),
            sa.ForeignKey("upload_batches.id", ondelete="CASCADE"),
            primary_key=True,
        ),
        sa.Column("asset_id", sa.String(), primary_key=True),
        sa.Column("path", sa.String(), nullable=False),
        # The bytes as they were when the batch was created; SHA-1 is what Immich
        # uses to recognise a file it already holds.
        sa.Column("sha256", sa.String(), nullable=True),
        sa.Column("sha1", sa.String(), nullable=True),
        sa.Column("state", sa.String(), nullable=False, server_default="pending"),
        sa.Column(
            "updated_at",
            sa.DateTime(timezone=True),
            nullable=False,
            server_default=sa.text("CURRENT_TIMESTAMP"),
        ),
    )
 def downgrade() -> None:
    op.drop_table("upload_items")
    op.drop_table("upload_batches")
--- a/photo_pipeline/main.py
+++ b/photo_pipeline/main.py
@@ -36,7 +36,7 @@ def main(argv: Sequence[str] | None = None) -> int:
        run_migrations(config.database_url)
        engine = create_db_engine(config.database_url)
-        Worker(create_session_factory(engine), worker_id=args.id, config=config).run_forever()
+        Worker(create_session_factory(engine), worker_id=args.id).run_forever()
        return 0
    import uvicorn
--- a/photo_pipeline/api/app.py
+++ b/photo_pipeline/api/app.py
@@ -25,7 +25,6 @@ from photo_pipeline.api.routes import (
    renames,
    safety,
    thumbnails,
    uploads,
    workflow,
 )
@@ -34,7 +33,6 @@ 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.logging import configure_logging
 from photo_pipeline.services.upload_batches import UploadBatchService
 FRONTEND_DIR = Path(__file__).resolve().parents[2] / "frontend"
@@ -51,9 +49,6 @@ def create_app(config: Config | None = None) -> FastAPI:
        app.state.config = config
        app.state.engine = engine
        app.state.session_factory = create_session_factory(engine)
        # An upload whose process died left no outcome behind; resolve it now so the
        # uploader lane is free and the uncertain batch is visible (US05-02).
        UploadBatchService(app.state.session_factory, config=config).recover()
        try:
            yield
        finally:
@@ -72,7 +67,6 @@ 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")
--- a/photo_pipeline/api/routes/uploads.py
+++ b/photo_pipeline/api/routes/uploads.py
@@ -1,113 +0,0 @@
 """Upload preflight and batch API (US05-01, US05-02).
 Preflight is a command, not a resource read: it contacts the Immich server, hashes
 the current bytes, and issues a token. Creating a batch requires that token, and
 starting one enqueues a durable job on the single ``upload`` lane — the API never
 runs the uploader in the request thread.
 """
 from __future__ import annotations
 from fastapi import APIRouter, Request
 from fastapi.responses import JSONResponse
 from pydantic import BaseModel
 from photo_pipeline.jobs.domain_handlers import UPLOAD_BATCH, UPLOAD_LOCK
 from photo_pipeline.services.jobs import JobBlocked, JobService
 from photo_pipeline.services.upload_batches import (
    BatchConflict,
    BatchError,
    UploadBatchService,
 )
 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
 class CreateBatchRequest(PreflightRequest):
    # The token of the preflight the user approved; a stale one is refused.
    token: str
 def _service(request: Request) -> UploadService:
    return UploadService(request.app.state.session_factory, config=request.app.state.config)
 def _batches(request: Request) -> UploadBatchService:
    return UploadBatchService(request.app.state.session_factory, config=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.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 _error(422, "unknown_album", str(error))
@router.post("/upload-batches", status_code=201)
 def create_batches(body: CreateBatchRequest, request: Request):
    """Turn an approved preflight into one durable batch per album."""
    try:
        return {"batches": _batches(request).create(
            body.albums, token=body.token, allow_partial=body.allow_partial
        )}
    except UploadError as error:
        return _error(422, "unknown_album", str(error))
    except BatchConflict as error:
        return _error(409, error.code, str(error))
    except BatchError as error:
        return _error(422, error.code, str(error))
@router.get("/upload-batches")
 def list_batches(request: Request) -> dict:
    return {"batches": _batches(request).list()}
@router.get("/upload-batches/{batch_id}")
 def get_batch(batch_id: str, request: Request):
    batch = _batches(request).get(batch_id)
    if batch is None:
        return _error(404, "not_found", f"unknown upload batch {batch_id}")
    return batch
@router.post("/upload-batches/{batch_id}/start")
 def start_batch(batch_id: str, request: Request):
    """Queue the batch on the uploader lane. The worker performs the upload."""
    service = _batches(request)
    batch = service.get(batch_id)
    if batch is None:
        return _error(404, "not_found", f"unknown upload batch {batch_id}")
    try:
        job = JobService(request.app.state.session_factory).enqueue(
            UPLOAD_BATCH,
            lock=UPLOAD_LOCK,
            # One queued attempt per batch attempt: a double-clicked start reuses it.
            idempotency_key=f"upload:{batch_id}:{batch['attempt_count']}",
            items=[batch_id],
        )
    except JobBlocked as error:
        return _error(409, error.code, str(error))
    return {"batch_id": batch_id, "job": job}
@router.post("/upload-batches/{batch_id}/cancel")
 def cancel_batch(batch_id: str, request: Request):
    try:
        return _batches(request).cancel(batch_id)
    except BatchError as error:
        return _error(404 if error.code == "not_found" else 409, error.code, str(error))
--- a/photo_pipeline/config.py
+++ b/photo_pipeline/config.py
@@ -37,8 +37,6 @@ 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:
--- a/photo_pipeline/integrations/immich_go.py
+++ b/photo_pipeline/integrations/immich_go.py
@@ -1,226 +0,0 @@
 """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.
 :func:`run_upload` is the only place the uploader is actually executed. It never
 uses a shell (the argument list goes straight to ``execve``, so no path or album
 name can be interpreted), streams the report to a file with a byte cap so a chatty
 or looping uploader cannot fill the disk, scrubs the API key out of anything the
 process echoes back, and polls a cancellation callback so a running upload can be
 stopped without killing the worker.
 """
 from __future__ import annotations
 import json
 import os
 import shutil
 import signal
 import subprocess
 import threading
 import time
 import urllib.error
 import urllib.request
 from collections.abc import Callable
 from pathlib import Path
 REDACTED = "***"
 PING_PATH = "/api/server/ping"
 PING_TIMEOUT_SECONDS = 5.0
 # Reports are kept in full up to this size; beyond it the tail is dropped and the
 # result is flagged truncated rather than growing without bound (concept §17).
 MAX_REPORT_BYTES = 4_000_000
 UPLOAD_TIMEOUT_SECONDS = 6 * 60 * 60
 POLL_SECONDS = 0.05
 # Grace period between asking the uploader to stop and killing it.
 TERMINATE_GRACE_SECONDS = 10.0
 # How long to wait for the last output after the process is gone. A child that
 # outlived its parent can still hold the pipe; the report is not worth hanging for.
 DRAIN_SECONDS = 2.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 run_upload(
    command: list[str],
    *,
    report_path: Path | str,
    secret: str | None = None,
    max_report_bytes: int | None = None,  # resolved at call time; see MAX_REPORT_BYTES
    cancelled: Callable[[], bool] | None = None,
    timeout: float = UPLOAD_TIMEOUT_SECONDS,
 ) -> dict:
    """Run one upload and return its outcome.
    ``{"exit_code", "cancelled", "timed_out", "report_path", "report_bytes",
    "report_truncated"}``. Output is streamed to ``report_path`` with ``secret``
    masked and the file capped at ``max_report_bytes``; the pipe keeps being drained
    after the cap so the child never blocks on a full buffer. ``cancelled`` is polled
    while the process runs: when it returns true the uploader is asked to stop, then
    killed if it does not.
    """
    report_path = Path(report_path)
    report_path.parent.mkdir(parents=True, exist_ok=True)
    max_report_bytes = MAX_REPORT_BYTES if max_report_bytes is None else max_report_bytes
    needle = (secret or "").encode() or None
    written = 0
    truncated = False
    process = subprocess.Popen(  # noqa: S603 — argv list, never a shell string
        command,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        shell=False,
        # Own process group: stopping the upload must stop whatever the uploader
        # spawned too, not leave orphans holding the pipe open.
        start_new_session=True,
    )
    def _drain() -> None:
        nonlocal written, truncated
        with open(report_path, "wb") as report:
            for line in process.stdout:  # line granularity keeps the mask reliable
                if needle:
                    line = line.replace(needle, REDACTED.encode())
                if written >= max_report_bytes:
                    truncated = True
                    continue  # keep draining the pipe, stop growing the file
                room = max_report_bytes - written
                report.write(line[:room])
                written += min(len(line), room)
                truncated = truncated or len(line) > room
                report.flush()
    reader = threading.Thread(target=_drain, daemon=True)
    reader.start()
    stopped = timed_out = False
    deadline = time.monotonic() + timeout
    while process.poll() is None:
        if cancelled is not None and cancelled():
            stopped = True
        elif time.monotonic() >= deadline:
            timed_out = True
        if stopped or timed_out:
            _stop(process)
            break
        time.sleep(POLL_SECONDS)
    exit_code = process.wait()
    reader.join(timeout=DRAIN_SECONDS)
    if process.stdout is not None:
        process.stdout.close()
    return {
        "exit_code": exit_code,
        "cancelled": stopped,
        "timed_out": timed_out,
        "report_path": str(report_path),
        "report_bytes": written,
        "report_truncated": truncated,
    }
 def _stop(process: subprocess.Popen) -> None:
    """Ask the uploader's whole process group to stop, then kill what remains."""
    _signal_group(process, signal.SIGTERM)
    try:
        process.wait(timeout=TERMINATE_GRACE_SECONDS)
    except subprocess.TimeoutExpired:
        _signal_group(process, signal.SIGKILL)
        process.wait()
 def _signal_group(process: subprocess.Popen, sig: int) -> None:
    try:
        os.killpg(os.getpgid(process.pid), sig)
    except (ProcessLookupError, PermissionError, OSError):
        # No group (already reaped, or a platform without them): signal the child.
        process.send_signal(sig)
 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,
        )
    )
--- a/photo_pipeline/jobs/domain_handlers.py
+++ b/photo_pipeline/jobs/domain_handlers.py
@@ -1,10 +1,10 @@
-"""Domain job handlers: safety scoring, content analysis, uploads (US02-06, US05-02).
+"""Domain job handlers: safety scoring and content analysis (US02-06).
-Importing this module registers the ``safety_score``, ``analysis``, and
+Importing this module registers the ``safety_score`` and ``analysis`` job types so
-``upload_batch`` job types so the generic worker can run them per item. Each handler
+the generic worker can run them per item (one item = one ``asset_id``). Each handler
 delegates to its service, which owns the real work and the privacy gate. Handlers
 are idempotent: re-scoring or re-analyzing one asset is safe after an interrupted
-attempt, and an upload batch refuses to re-run an attempt whose outcome is unknown.
+attempt.
 Providers/models are the service defaults here (real NsfwModel / vision provider);
 tests exercise the services directly with injected fakes rather than the worker.
@@ -12,15 +12,12 @@ tests exercise the services directly with injected fakes rather than the worker.
 from __future__ import annotations
-from photo_pipeline.jobs.handlers import Cancelled, JobContext, register
+from photo_pipeline.jobs.handlers import JobContext, register
 SAFETY_SCORE = "safety_score"
 ANALYSIS = "analysis"
 UPLOAD_BATCH = "upload_batch"
 # Both mutate the library's metadata/derived state; one at a time (concept §one job).
 LIBRARY_WRITE_LOCK = "library_write"
 # The uploader lane: one album batch at a time (concept §16).
 UPLOAD_LOCK = "upload"
 def _safety_score_item(asset_id: str, ctx: JobContext) -> None:
@@ -35,22 +32,5 @@ def _analysis_item(asset_id: str, ctx: JobContext) -> None:
    AnalysisService(ctx.session_factory).run([asset_id])
 def _upload_batch_item(batch_id: str, ctx: JobContext) -> None:
    """One item = one album batch. The upload itself is long and external, so the
    handler hands the job's cancellation check to the service, which stops the
    uploader and leaves a resumable batch."""
    from photo_pipeline.config import Config
    from photo_pipeline.services.upload_batches import BatchState, UploadBatchService
    config = ctx.config if ctx.config is not None else Config.from_env()
    service = UploadBatchService(ctx.session_factory, config=config)
    batch = service.run(batch_id, worker_id=ctx.worker_id, cancelled=ctx.cancelled)
    if batch["state"] == BatchState.CANCELLED:
        raise Cancelled(f"upload batch {batch_id} was cancelled")
    if batch["state"] in (BatchState.FAILED, BatchState.UNKNOWN):
        raise RuntimeError(f"upload batch {batch_id} is {batch['state']}: {batch['error_code']}")
 register(SAFETY_SCORE, _safety_score_item)
 register(ANALYSIS, _analysis_item)
 register(UPLOAD_BATCH, _upload_batch_item)
--- a/photo_pipeline/jobs/handlers.py
+++ b/photo_pipeline/jobs/handlers.py
@@ -34,9 +34,6 @@ class JobContext:
    fencing_token: int
    service: "JobService"
    session_factory: object | None = None
    # Handlers that talk to an external tool (uploads) need the typed configuration;
    # the worker passes its own so a test stack is never read from the environment.
    config: object | None = None
    def cancelled(self) -> bool:
        from photo_pipeline.services.jobs import JobState
--- a/photo_pipeline/jobs/worker.py
+++ b/photo_pipeline/jobs/worker.py
@@ -30,7 +30,6 @@ class Worker:
        *,
        job_types: Sequence[str] | None = None,
        lease_seconds: int = 60,
        config: object | None = None,
    ) -> None:
        self._session_factory = session_factory
        self.service = JobService(session_factory)
@@ -38,7 +37,6 @@ class Worker:
        self.worker_id = worker_id
        self.job_types = list(job_types if job_types is not None else self.handlers.keys())
        self.lease_seconds = lease_seconds
        self.config = config
    def run_once(self) -> str | None:
        """Recover stragglers, then claim and fully process one job. Returns its id."""
@@ -53,9 +51,7 @@ class Worker:
    def _process(self, job_id: str, job_type: str, token: int) -> None:
        handler = self.handlers[job_type]
-        ctx = JobContext(
+        ctx = JobContext(job_id, self.worker_id, token, self.service, self._session_factory)
            job_id, self.worker_id, token, self.service, self._session_factory, self.config
        )
        self._reset_interrupted_items(job_id, token)
        cancelled = False
--- a/photo_pipeline/models/init.py
+++ b/photo_pipeline/models/init.py
@@ -14,7 +14,6 @@ from photo_pipeline.models.duplicates import (
 from photo_pipeline.models.jobs import Job, JobEvent, JobItem
 from photo_pipeline.models.renames import RenameOperation, RenamePlan
 from photo_pipeline.models.thumbnails import Thumbnail
 from photo_pipeline.models.uploads import UploadBatch, UploadItem
 from photo_pipeline.models.workflow import AnalysisResult, SafetyReview
 __all__ = [
@@ -30,8 +29,6 @@ __all__ = [
    "RenamePlan",
    "RenameOperation",
    "Thumbnail",
    "UploadBatch",
    "UploadItem",
    "SafetyReview",
    "AnalysisResult",
 ]
--- a/photo_pipeline/models/uploads.py
+++ b/photo_pipeline/models/uploads.py
@@ -1,90 +0,0 @@
 """Upload batch persistence (US05-02).
 One batch is one approved album folder handed to ``immich-go``. It is the durable
 record of an irreversible external action, so it stores everything needed to answer
 "which exact bytes did we send, with which command, and how did it end?" after a
 crash: the preflight token that authorised it, the redacted command, the uploader
 version, the per-asset pre-upload hashes, every attempt, and where the raw report
 was written.
 Item rows keep both digests: SHA-256 is the app's byte identity and SHA-1 is what
 Immich/immich-go use to recognise a file it already has (concept §8).
 """
 from __future__ import annotations
 from datetime import datetime
 from sqlalchemy import (
    Boolean,
    DateTime,
    ForeignKey,
    Integer,
    String,
    func,
 )
 from sqlalchemy.orm import Mapped, mapped_column
 from photo_pipeline.db import Base
 class UploadBatch(Base):
    __tablename__ = "upload_batches"
    id: Mapped[str] = mapped_column(String, primary_key=True)
    album: Mapped[str] = mapped_column(String, nullable=False, index=True)
    folder: Mapped[str] = mapped_column(String, nullable=False)
    album_name: Mapped[str] = mapped_column(String, nullable=False)
    # planned | running | cancelling | cancelled | succeeded | failed
    # | unknown_requires_verification
    state: Mapped[str] = mapped_column(String, nullable=False, default="planned")
    # The preflight token this batch was approved against; re-checked before every
    # attempt so changed bytes or decisions cannot be uploaded silently.
    preflight_token: Mapped[str] = mapped_column(String, nullable=False)
    allow_partial: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
    command: Mapped[str | None] = mapped_column(String)  # JSON array, redacted
    uploader_version: Mapped[str | None] = mapped_column(String)
    asset_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
    attempt_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
    # Bumped on every claim and used as the fencing token, so a superseded attempt
    # cannot commit its outcome.
    version: Mapped[int] = mapped_column(Integer, nullable=False, default=1)
    worker_id: Mapped[str | None] = mapped_column(String)
    report_path: Mapped[str | None] = mapped_column(String)
    report_bytes: Mapped[int | None] = mapped_column(Integer)
    report_truncated: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
    exit_code: Mapped[int | None] = mapped_column(Integer)
    error_code: Mapped[str | None] = mapped_column(String)
    error_message: Mapped[str | None] = mapped_column(String)
    started_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    finished_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now()
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now()
    )
 class UploadItem(Base):
    __tablename__ = "upload_items"
    batch_id: Mapped[str] = mapped_column(
        ForeignKey("upload_batches.id", ondelete="CASCADE"), primary_key=True
    )
    asset_id: Mapped[str] = mapped_column(String, primary_key=True)
    path: Mapped[str] = mapped_column(String, nullable=False)
    # Hashes of the bytes as they were when the batch was created.
    sha256: Mapped[str | None] = mapped_column(String)
    sha1: Mapped[str | None] = mapped_column(String)
    # pending | sent | failed — the per-asset upload *result* is parsed from the
    # report in US05-03 and verified in US05-04; ``sent`` only means the batch
    # process exited successfully.
    state: Mapped[str] = mapped_column(String, nullable=False, default="pending")
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now()
    )
--- a/photo_pipeline/services/hashing.py
+++ b/photo_pipeline/services/hashing.py
@@ -24,16 +24,7 @@ _CHUNK = 1 << 20
 def sha256_file(path: Path | str) -> str:
-    return _digest_file(path, hashlib.sha256())
+    digest = hashlib.sha256()
 def sha1_file(path: Path | str) -> str:
    """SHA-1 of the file bytes. Not an identity hash here — it is the checksum
    Immich/immich-go use to recognise an asset they already hold (concept §8)."""
    return _digest_file(path, hashlib.sha1())
 def _digest_file(path: Path | str, digest) -> str:
    with open(path, "rb") as handle:
        for chunk in iter(lambda: handle.read(_CHUNK), b""):
            digest.update(chunk)
--- a/photo_pipeline/services/upload_batches.py
+++ b/photo_pipeline/services/upload_batches.py
@@ -1,424 +0,0 @@
 """UploadBatchService — one approved album at a time through immich-go (US05-02).
 Preflight (US05-01) proves a scope is safe and issues a token; this service turns
 that approval into a durable batch and runs it. Upload is the one stage the app
 cannot undo, so the discipline is:
 - **the approval is re-proved, not remembered.** Before every attempt the batch's
  preflight token is recomputed from the current library state. Bytes edited after
  approval, a withdrawn safety decision, or a server that stopped answering all
  produce a different token and the attempt is refused, never run "optimistically".
 - **one lane.** A batch can only start while no other batch is running; the album
  scope of a batch never widens after creation (concept §16 uploader lane).
 - **cancellation is cooperative and durable.** ``cancel`` writes ``cancelling``;
  the running attempt observes it through the database, stops the uploader, and
  records ``cancelled``. A cancelled batch is re-runnable from its own boundary.
 - **an interrupted attempt is uncertain, not failed.** Immich may have accepted
  files the app never saw a report for, so ``recover`` marks a batch whose worker
  vanished ``unknown_requires_verification`` (concept §15) instead of retrying it
  blindly. Resolving that is US05-04.
 Per-asset upload *results* are deliberately not interpreted here — parsing the
 report is US05-03. What this story guarantees is that the report exists, is bounded,
 carries no credential, and that the batch's state is an honest description of what
 the process did.
 """
 from __future__ import annotations
 import json
 import uuid
 from datetime import datetime, timezone
 from pathlib import Path
 from sqlalchemy import select, update
 from sqlalchemy.orm import sessionmaker
 from photo_pipeline.config import Config
 from photo_pipeline.integrations import immich_go
 from photo_pipeline.models import UploadBatch, UploadItem
 from photo_pipeline.services.hashing import sha1_file
 from photo_pipeline.services.uploads import UploadService
 class BatchState:
    PLANNED = "planned"
    RUNNING = "running"
    CANCELLING = "cancelling"
    CANCELLED = "cancelled"
    SUCCEEDED = "succeeded"
    FAILED = "failed"
    # The attempt died without a parsed outcome: the server may hold the files.
    UNKNOWN = "unknown_requires_verification"
 # States that occupy the single uploader lane.
 LANE_STATES = frozenset({BatchState.RUNNING, BatchState.CANCELLING})
 # States a batch may be (re)started from. ``unknown_requires_verification`` is not
 # among them: an uncertain upload must be verified (US05-04), never blindly retried.
 RUNNABLE_STATES = frozenset({BatchState.PLANNED, BatchState.FAILED, BatchState.CANCELLED})
 # States whose batch is still the live one for its album.
 OPEN_STATES = frozenset({BatchState.PLANNED, BatchState.RUNNING, BatchState.CANCELLING})
 class ItemState:
    PENDING = "pending"
    # ``sent`` means the batch process exited cleanly, not that Immich confirmed the
    # asset — the per-item outcome comes from the report in US05-03/US05-04.
    SENT = "sent"
    FAILED = "failed"
 class BatchError(RuntimeError):
    """The request cannot be carried out (unknown batch, wrong state, blocked)."""
    def __init__(self, code: str, message: str) -> None:
        super().__init__(message)
        self.code = code
 class BatchConflict(BatchError):
    """The lane is busy or the approval is stale — retry after resolving it."""
 def _now() -> datetime:
    return datetime.now(timezone.utc)
 class UploadBatchService:
    def __init__(self, session_factory: sessionmaker, *, config: Config) -> None:
        self._session_factory = session_factory
        self._config = config
        self._preflight = UploadService(session_factory, config=config)
    # ── creation ──────────────────────────────────────────────────────────────
    def create(
        self, albums: list[str] | None = None, *, token: str, allow_partial: bool = False
    ) -> list[dict]:
        """Turn a *ready* preflight into one durable batch per album.
        ``token`` must be the token of the current preflight for the same scope and
        policy; anything else means the browser is acting on a stale preview.
        """
        report = self._preflight.preflight(albums, allow_partial=allow_partial)
        if token != report["token"]:
            raise BatchConflict(
                "stale_preflight", "the preflight token does not describe the current state"
            )
        if report["state"] != "ready":
            raise BatchError("not_ready", "the scope has unresolved upload blockers")
        created: list[dict] = []
        for album in report["albums"]:
            existing = self._open_batch_for(album["album"])
            if existing is not None:
                created.append(existing)  # idempotent: one open batch per album
                continue
            created.append(self._create_one(album, token=token, allow_partial=allow_partial))
        return created
    def _create_one(self, album: dict, *, token: str, allow_partial: bool) -> dict:
        eligible = [asset for asset in album["assets"] if not asset["blockers"]]
        batch_id = str(uuid.uuid4())
        command = immich_go.preview_command(
            binary=self._config.immich_go_binary,
            server_url=self._config.immich_server_url,
            album_name=album["album_name"],
            folder=album["folder"],
        )
        with self._session_factory() as session:
            session.add(
                UploadBatch(
                    id=batch_id,
                    album=album["album"],
                    folder=album["folder"],
                    album_name=album["album_name"],
                    state=BatchState.PLANNED,
                    preflight_token=token,
                    allow_partial=allow_partial,
                    command=json.dumps(command),
                    uploader_version=immich_go.version(self._config.immich_go_binary),
                    asset_count=len(eligible),
                )
            )
            for asset in eligible:
                session.add(
                    UploadItem(
                        batch_id=batch_id,
                        asset_id=asset["asset_id"],
                        path=asset["current_path"],
                        sha256=asset["current_sha256"],
                        sha1=sha1_file(asset["current_path"]),
                        state=ItemState.PENDING,
                    )
                )
            session.commit()
        return self.get(batch_id)
    # ── running ───────────────────────────────────────────────────────────────
    def run(self, batch_id: str, *, worker_id: str = "uploader", cancelled=None) -> dict:
        """Run one batch to completion. Blocks for the duration of the upload."""
        batch = self._require(batch_id)
        if batch["state"] not in RUNNABLE_STATES:
            code = (
                "requires_verification"
                if batch["state"] == BatchState.UNKNOWN
                else "not_runnable"
            )
            raise BatchError(code, f"batch {batch_id} is {batch['state']}")
        busy = [row for row in self.list() if row["id"] != batch_id and row["state"] in LANE_STATES]
        if busy:
            raise BatchConflict("lane_busy", f"upload batch {busy[0]['id']} is still running")
        # The approval is re-proved here, immediately before the irreversible act.
        if not self._preflight.verify_token(
            batch["preflight_token"], [batch["album"]], allow_partial=batch["allow_partial"]
        ):
            self._finish(
                batch_id,
                token=batch["version"],
                state=BatchState.FAILED,
                error=("stale_preflight", "the library changed after this batch was approved"),
            )
            raise BatchConflict(
                "stale_preflight", "the library changed after this batch was approved"
            )
        token = self._claim(batch_id, worker_id=worker_id)
        attempt = self.get(batch_id)["attempt_count"]
        report_path = (
            Path(self._config.data_dir) / "uploads" / f"{batch_id}-attempt-{attempt}.log"
        )
        key = self._config.immich_api_key.get_secret_value() if self._config.immich_api_key else ""
        command = immich_go.build_command(
            binary=self._config.immich_go_binary,
            server_url=self._config.immich_server_url,
            api_key=key,
            album_name=batch["album_name"],
            folder=batch["folder"],
        )
        def _stop_requested() -> bool:
            if cancelled is not None and cancelled():
                return True
            current = self.get(batch_id)
            return current is None or current["state"] == BatchState.CANCELLING
        try:
            result = immich_go.run_upload(
                command,
                report_path=report_path,
                secret=key or None,
                cancelled=_stop_requested,
            )
        except OSError as error:  # uploader vanished between preflight and exec
            self._finish(
                batch_id,
                token=token,
                state=BatchState.FAILED,
                error=("uploader_failed", str(error)),
            )
            return self.get(batch_id)
        if result["cancelled"]:
            state, error, item_state = BatchState.CANCELLED, None, None
        elif result["timed_out"]:
            # Killed mid-flight: the server may already hold some of the files.
            state = BatchState.UNKNOWN
            error = ("timeout", "the uploader exceeded its time limit and was stopped")
            item_state = None
        elif result["exit_code"] == 0:
            state, error, item_state = BatchState.SUCCEEDED, None, ItemState.SENT
        else:
            state = BatchState.FAILED
            error = ("uploader_failed", f"immich-go exited with {result['exit_code']}")
            item_state = ItemState.FAILED
        self._finish(batch_id, token=token, state=state, error=error, result=result)
        if item_state:
            self._set_items(batch_id, item_state)
        return self.get(batch_id)
    def cancel(self, batch_id: str) -> dict:
        """Request a stop. A running attempt drains; a planned batch stops outright."""
        batch = self._require(batch_id)
        if batch["state"] == BatchState.PLANNED:
            target = BatchState.CANCELLED
        elif batch["state"] == BatchState.RUNNING:
            target = BatchState.CANCELLING
        else:
            raise BatchError("not_cancellable", f"batch {batch_id} is {batch['state']}")
        with self._session_factory() as session:
            row = session.get(UploadBatch, batch_id)
            row.state = target
            row.updated_at = _now()
            if target == BatchState.CANCELLED:
                row.finished_at = _now()
            session.commit()
        return self.get(batch_id)
    # ── recovery ──────────────────────────────────────────────────────────────
    def recover(self) -> dict:
        """Resolve batches whose attempt died with the process.
        A batch that never started is left ``planned`` and simply runs later. One
        that was mid-upload cannot be classified from local state — immich-go may
        have transferred everything before the crash — so it becomes
        ``unknown_requires_verification`` and releases the lane rather than being
        retried or declared failed.
        """
        interrupted = 0
        with self._session_factory() as session:
            for row in session.scalars(
                select(UploadBatch).where(UploadBatch.state.in_(LANE_STATES))
            ):
                row.state = BatchState.UNKNOWN
                row.error_code = "interrupted"
                row.error_message = "the uploader process ended without a recorded outcome"
                row.finished_at = _now()
                row.updated_at = _now()
                row.version += 1
                interrupted += 1
            session.commit()
        return {"interrupted": interrupted}
    # ── reads ─────────────────────────────────────────────────────────────────
    def get(self, batch_id: str) -> dict | None:
        with self._session_factory() as session:
            row = session.get(UploadBatch, batch_id)
            if row is None:
                return None
            items = list(
                session.scalars(
                    select(UploadItem)
                    .where(UploadItem.batch_id == batch_id)
                    .order_by(UploadItem.path)
                )
            )
            return _batch_dict(row, items)
    def list(self) -> list[dict]:
        with self._session_factory() as session:
            rows = list(session.scalars(select(UploadBatch).order_by(UploadBatch.created_at)))
            return [_batch_dict(row, []) for row in rows]
    def report(self, batch_id: str) -> str:
        """The raw uploader output kept for this batch, or ``""`` when there is none."""
        batch = self._require(batch_id)
        path = Path(batch["report_path"]) if batch["report_path"] else None
        if path is None or not path.exists():
            return ""
        return path.read_text(errors="replace")
    # ── internals ─────────────────────────────────────────────────────────────
    def _open_batch_for(self, album: str) -> dict | None:
        with self._session_factory() as session:
            row = session.scalar(
                select(UploadBatch).where(
                    UploadBatch.album == album, UploadBatch.state.in_(OPEN_STATES)
                )
            )
            return self.get(row.id) if row else None
    def _claim(self, batch_id: str, *, worker_id: str) -> int:
        """Take ownership: bump the version (the fencing token) and start an attempt."""
        with self._session_factory() as session:
            row = session.get(UploadBatch, batch_id)
            row.version += 1
            row.attempt_count += 1
            row.state = BatchState.RUNNING
            row.worker_id = worker_id
            row.error_code = row.error_message = None
            row.started_at = _now()
            row.finished_at = None
            row.updated_at = _now()
            token = row.version
            session.commit()
        return token
    def _finish(
        self,
        batch_id: str,
        *,
        token: int,
        state: str,
        error: tuple[str, str] | None = None,
        result: dict | None = None,
    ) -> None:
        """Record the outcome, but only for the attempt that still owns the batch."""
        values = {
            "state": state,
            "finished_at": _now(),
            "updated_at": _now(),
            "error_code": error[0] if error else None,
            "error_message": error[1][:500] if error else None,
        }
        if result is not None:
            values |= {
                "exit_code": result["exit_code"],
                "report_path": result["report_path"],
                "report_bytes": result["report_bytes"],
                "report_truncated": result["report_truncated"],
            }
        with self._session_factory() as session:
            session.execute(
                update(UploadBatch)
                .where(UploadBatch.id == batch_id, UploadBatch.version == token)
                .values(**values)
            )
            session.commit()
    def _set_items(self, batch_id: str, state: str) -> None:
        with self._session_factory() as session:
            session.execute(
                update(UploadItem)
                .where(UploadItem.batch_id == batch_id)
                .values(state=state, updated_at=_now())
            )
            session.commit()
    def _require(self, batch_id: str) -> dict:
        batch = self.get(batch_id)
        if batch is None:
            raise BatchError("not_found", f"unknown upload batch {batch_id!r}")
        return batch
 def _batch_dict(row: UploadBatch, items: list[UploadItem]) -> dict:
    return {
        "id": row.id,
        "album": row.album,
        "folder": row.folder,
        "album_name": row.album_name,
        "state": row.state,
        "preflight_token": row.preflight_token,
        "allow_partial": row.allow_partial,
        "command": json.loads(row.command or "[]"),
        "uploader_version": row.uploader_version,
        "asset_count": row.asset_count,
        "attempt_count": row.attempt_count,
        "version": row.version,
        "worker_id": row.worker_id,
        "report_path": row.report_path,
        "report_bytes": row.report_bytes,
        "report_truncated": row.report_truncated,
        "exit_code": row.exit_code,
        "error_code": row.error_code,
        "error_message": row.error_message,
        "started_at": row.started_at.isoformat() if row.started_at else None,
        "finished_at": row.finished_at.isoformat() if row.finished_at else None,
        "items": [
            {
                "asset_id": item.asset_id,
                "path": item.path,
                "sha256": item.sha256,
                "sha1": item.sha1,
                "state": item.state,
            }
            for item in items
        ],
    }
--- a/photo_pipeline/services/uploads.py
+++ b/photo_pipeline/services/uploads.py
@@ -1,306 +0,0 @@
 """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}"
--- a/tests/integration/test_upload_batches.py
+++ b/tests/integration/test_upload_batches.py
@@ -1,588 +0,0 @@
 """Upload batch orchestration (US05-02).
 The uploader is a real executable on disk driven through the real integration
 adapter and ``subprocess`` — never a mock — so argument construction, output
 bounding, credential privacy, and killing a running process are exercised the way
 production does them. Each fake uploader records the argv it was given into a
 side-channel file, which is what the argument/album-isolation assertions read.
 """
 import json
 import os
 import signal
 import stat
 import threading
 import time
 import uuid
 from datetime import datetime, timezone
 from http.server import BaseHTTPRequestHandler, HTTPServer
 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.jobs.domain_handlers import UPLOAD_BATCH
 from photo_pipeline.jobs.worker import Worker
 from photo_pipeline.models import AnalysisResult, Asset, SafetyReview, UploadBatch
 from photo_pipeline.services.hashing import sha1_file, sha256_file
 from photo_pipeline.services.jobs import JobService, JobState
 from photo_pipeline.services.upload_batches import (
    BatchConflict,
    BatchError,
    BatchState,
    ItemState,
    UploadBatchService,
 )
 from photo_pipeline.services.uploads import UploadService
 NOW = datetime(2026, 1, 1, tzinfo=timezone.utc)
 SENTINEL_KEY = "immich-sentinel-9f3a2b"
 UPLOADER_VERSION = "immich-go 0.21.0"
 # ── fake external boundary ───────────────────────────────────────────────────
 class _PingHandler(BaseHTTPRequestHandler):
    def do_GET(self):  # noqa: N802 (BaseHTTPRequestHandler API)
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.end_headers()
        self.wfile.write(b'{"res":"pong"}')
    def log_message(self, *args):
        pass
@pytest.fixture
 def immich_server():
    server = HTTPServer(("127.0.0.1", 0), _PingHandler)
    threading.Thread(target=server.serve_forever, daemon=True).start()
    yield f"http://127.0.0.1:{server.server_port}"
    server.shutdown()
    server.server_close()
 def _uploader(tmp_path, body: str = "", *, name="immich-go"):
    """A real executable standing in for immich-go.
    ``--version`` answers like the real tool; any other invocation appends its
    complete argv to ``<name>.argv`` and then runs ``body``.
    """
    path = tmp_path / name
    argv_log = tmp_path / f"{name}.argv"
    path.write_text(
        "#!/bin/sh\n"
        f'if [ "$1" = "--version" ]; then echo "{UPLOADER_VERSION}"; exit 0; fi\n'
        f'printf "%s\\n" "$*" >> "{argv_log}"\n'
        f"{body}\n"
    )
    path.chmod(path.stat().st_mode | stat.S_IEXEC | stat.S_IXGRP | stat.S_IXOTH)
    return path
 def _argv(tmp_path, name="immich-go") -> list[str]:
    log = tmp_path / f"{name}.argv"
    return log.read_text().splitlines() if log.exists() else []
 # ── environment ──────────────────────────────────────────────────────────────
 def _env(tmp_path, server_url, *, uploader=None):
    (tmp_path / "data").mkdir(exist_ok=True)
    lib = tmp_path / "lib"
    lib.mkdir(exist_ok=True)
    config = Config.from_env(
        {
            "PHOTO_PIPELINE_DATA_DIR": str(tmp_path / "data"),
            "PHOTO_PIPELINE_LIBRARY_ROOTS": str(lib),
            "PHOTO_PIPELINE_IMMICH_SERVER_URL": server_url,
            "PHOTO_PIPELINE_IMMICH_API_KEY": SENTINEL_KEY,
            "PHOTO_PIPELINE_IMMICH_GO_BINARY": str(
                uploader if uploader is not None else _uploader(tmp_path, "exit 0")
            ),
        }
    )
    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")):
    folder = lib / album
    folder.mkdir(parents=True, exist_ok=True)
    with sf() as session:
        for name in names:
            path = folder / name
            path.write_bytes(f"{album}/{name}".encode() * 16)
            asset_id = str(uuid.uuid4())
            session.add(
                Asset(
                    id=asset_id,
                    original_path=str(path),
                    current_path=str(path),
                    discovered_at=NOW,
                    hash_version=1,
                    byte_size=path.stat().st_size,
                    current_sha256=sha256_file(path),
                )
            )
            session.add(
                SafetyReview(
                    id=str(uuid.uuid4()),
                    asset_id=asset_id,
                    decision="sfw",
                    exif_verified_at=NOW,
                )
            )
            session.add(
                AnalysisResult(asset_id=asset_id, status="analyzed", exif_written_at=NOW)
            )
        session.commit()
    return folder
 def _service(sf, config):
    return UploadBatchService(sf, config=config)
 def _approved(sf, config, albums=None, **kwargs):
    """Create batches from a fresh, ready preflight."""
    report = UploadService(sf, config=config).preflight(albums, **kwargs)
    assert report["state"] == "ready", report["blockers"]
    return _service(sf, config).create(albums, token=report["token"], **kwargs)
 # ── creation ─────────────────────────────────────────────────────────────────
 def test_batch_records_album_assets_hashes_and_command(tmp_path, immich_server):
    config, sf, lib = _env(tmp_path, immich_server)
    folder = _album(sf, lib)
    (batch,) = _approved(sf, config)
    assert batch["state"] == BatchState.PLANNED
    assert batch["album"] == "rome" and batch["folder"] == str(folder)
    assert batch["asset_count"] == 2 and len(batch["items"]) == 2
    assert batch["uploader_version"] == UPLOADER_VERSION
    assert "--album-name=rome" in batch["command"]
    assert "--api-key=***" in batch["command"]
    for item in batch["items"]:
        assert item["sha256"] == sha256_file(item["path"])
        assert item["sha1"] == sha1_file(item["path"])
        assert item["state"] == ItemState.PENDING
 def test_one_batch_per_album_and_scope_is_isolated(tmp_path, immich_server):
    config, sf, lib = _env(tmp_path, immich_server)
    _album(sf, lib, "rome")
    _album(sf, lib, "paris", names=("c.jpg",))
    batches = _approved(sf, config, ["paris"])
    assert [b["album"] for b in batches] == ["paris"]
    assert [b["asset_count"] for b in batches] == [1]
 def test_creating_twice_reuses_the_open_batch(tmp_path, immich_server):
    config, sf, lib = _env(tmp_path, immich_server)
    _album(sf, lib)
    first = _approved(sf, config)[0]
    second = _approved(sf, config)[0]
    assert first["id"] == second["id"]
 def test_stale_preflight_token_cannot_create_a_batch(tmp_path, immich_server):
    config, sf, lib = _env(tmp_path, immich_server)
    folder = _album(sf, lib)
    token = UploadService(sf, config=config).preflight()["token"]
    (folder / "a.jpg").write_bytes(b"edited after approval")
    with pytest.raises(BatchConflict) as error:
        _service(sf, config).create(token=token)
    assert error.value.code == "stale_preflight"
 def test_blocked_scope_cannot_create_a_batch(tmp_path, immich_server):
    config, sf, lib = _env(tmp_path, immich_server)
    folder = _album(sf, lib)
    (folder / "a.jpg").write_bytes(b"edited before approval")
    report = UploadService(sf, config=config).preflight()
    with pytest.raises(BatchError) as error:
        _service(sf, config).create(token=report["token"])
    assert error.value.code == "not_ready"
 # ── running the uploader ─────────────────────────────────────────────────────
 def test_run_invokes_the_uploader_with_the_batch_album_and_folder(tmp_path, immich_server):
    config, sf, lib = _env(tmp_path, immich_server)
    folder = _album(sf, lib)
    (batch,) = _approved(sf, config)
    result = _service(sf, config).run(batch["id"])
    assert result["state"] == BatchState.SUCCEEDED and result["exit_code"] == 0
    (invocation,) = _argv(tmp_path)
    assert "upload from-folder" in invocation
    assert "--album-name=rome" in invocation
    assert str(folder) in invocation
    assert all(item["state"] == ItemState.SENT for item in result["items"])
    assert result["attempt_count"] == 1 and result["started_at"] and result["finished_at"]
 def test_other_albums_are_never_passed_to_the_uploader(tmp_path, immich_server):
    config, sf, lib = _env(tmp_path, immich_server)
    _album(sf, lib, "rome")
    paris = _album(sf, lib, "paris", names=("c.jpg",))
    (batch,) = _approved(sf, config, ["rome"])
    _service(sf, config).run(batch["id"])
    (invocation,) = _argv(tmp_path)
    assert str(paris) not in invocation and "paris" not in invocation
 def test_album_names_are_arguments_not_shell_text(tmp_path, immich_server):
    """A folder whose name contains shell metacharacters must reach the uploader
    verbatim; nothing may be interpreted (no shell is involved)."""
    config, sf, lib = _env(tmp_path, immich_server)
    hostile = "rome; touch pwned"
    _album(sf, lib, hostile, names=("a.jpg",))
    (batch,) = _approved(sf, config)
    result = _service(sf, config).run(batch["id"])
    assert result["state"] == BatchState.SUCCEEDED
    assert f"--album-name={hostile}" in _argv(tmp_path)[0]
    assert not (Path.cwd() / "pwned").exists() and not (lib / "pwned").exists()
 def test_uploader_failure_is_recorded_and_items_are_not_sent(tmp_path, immich_server):
    config, sf, lib = _env(
        tmp_path, immich_server, uploader=_uploader(tmp_path, "echo 'boom' >&2; exit 3")
    )
    _album(sf, lib)
    (batch,) = _approved(sf, config)
    result = _service(sf, config).run(batch["id"])
    assert result["state"] == BatchState.FAILED and result["exit_code"] == 3
    assert result["error_code"] == "uploader_failed"
    assert all(item["state"] == ItemState.FAILED for item in result["items"])
    assert "boom" in _service(sf, config).report(batch["id"])
 def test_a_failed_batch_can_be_retried_as_a_new_attempt(tmp_path, immich_server):
    config, sf, lib = _env(tmp_path, immich_server, uploader=_uploader(tmp_path, "exit 3"))
    _album(sf, lib)
    (batch,) = _approved(sf, config)
    service = _service(sf, config)
    service.run(batch["id"])
    retried = service.run(batch["id"])
    assert retried["attempt_count"] == 2
    assert len(_argv(tmp_path)) == 2
 def test_bytes_edited_after_approval_block_the_upload(tmp_path, immich_server):
    config, sf, lib = _env(tmp_path, immich_server)
    folder = _album(sf, lib)
    (batch,) = _approved(sf, config)
    (folder / "a.jpg").write_bytes(b"edited after approval")
    with pytest.raises(BatchConflict) as error:
        _service(sf, config).run(batch["id"])
    assert error.value.code == "stale_preflight"
    assert _argv(tmp_path) == [], "the uploader must not run on changed bytes"
    assert _service(sf, config).get(batch["id"])["state"] == BatchState.FAILED
 # ── credential privacy ───────────────────────────────────────────────────────
 def test_the_api_key_reaches_the_uploader_but_never_the_record(tmp_path, immich_server):
    config, sf, lib = _env(tmp_path, immich_server)
    _album(sf, lib)
    (batch,) = _approved(sf, config)
    result = _service(sf, config).run(batch["id"])
    assert f"--api-key={SENTINEL_KEY}" in _argv(tmp_path)[0], "the real key must be passed"
    assert SENTINEL_KEY not in json.dumps(result, default=str)
 def test_a_key_echoed_by_the_uploader_is_scrubbed_from_the_report(tmp_path, immich_server):
    config, sf, lib = _env(
        tmp_path, immich_server, uploader=_uploader(tmp_path, 'echo "using key $4"; exit 0')
    )
    _album(sf, lib)
    (batch,) = _approved(sf, config)
    service = _service(sf, config)
    service.run(batch["id"])
    report = service.report(batch["id"])
    assert SENTINEL_KEY not in report and "***" in report
 # ── bounded output ───────────────────────────────────────────────────────────
 def test_a_chatty_uploader_cannot_grow_the_report_without_bound(tmp_path, immich_server):
    config, sf, lib = _env(
        tmp_path,
        immich_server,
        uploader=_uploader(tmp_path, "i=0; while [ $i -lt 2000 ]; do echo line-$i; i=$((i+1)); done"),
    )
    _album(sf, lib)
    (batch,) = _approved(sf, config)
    from photo_pipeline.integrations import immich_go
    service = _service(sf, config)
    original = immich_go.MAX_REPORT_BYTES
    immich_go.MAX_REPORT_BYTES = 500  # a cap small enough to hit in one test
    try:
        result = service.run(batch["id"])
    finally:
        immich_go.MAX_REPORT_BYTES = original
    assert result["state"] == BatchState.SUCCEEDED, "the uploader still finished normally"
    assert result["report_truncated"] is True
    assert result["report_bytes"] <= 500
    assert Path(result["report_path"]).stat().st_size <= 500
 # ── one lane ─────────────────────────────────────────────────────────────────
 def test_a_second_batch_cannot_run_while_one_is_running(tmp_path, immich_server):
    config, sf, lib = _env(tmp_path, immich_server)
    _album(sf, lib, "rome")
    _album(sf, lib, "paris", names=("c.jpg",))
    rome, paris = sorted(_approved(sf, config), key=lambda b: b["album"] != "rome")
    with sf() as session:  # a batch already occupying the lane
        session.get(UploadBatch, rome["id"]).state = BatchState.RUNNING
        session.commit()
    with pytest.raises(BatchConflict) as error:
        _service(sf, config).run(paris["id"])
    assert error.value.code == "lane_busy"
    assert _argv(tmp_path) == []
 def test_the_job_lock_refuses_a_second_queued_upload(tmp_path, immich_server):
    config, sf, lib = _env(tmp_path, immich_server)
    _album(sf, lib, "rome")
    _album(sf, lib, "paris", names=("c.jpg",))
    batches = _approved(sf, config)
    with TestClient(create_app(config)) as client:
        first = client.post(f"/api/v1/upload-batches/{batches[0]['id']}/start")
        second = client.post(f"/api/v1/upload-batches/{batches[1]['id']}/start")
    assert first.status_code == 200
    assert second.status_code == 409 and second.json()["error"]["code"] == "lock_held"
 # ── cancellation ─────────────────────────────────────────────────────────────
 def test_cancelling_a_running_batch_stops_the_uploader(tmp_path, immich_server):
    """The uploader sleeps; cancelling flips the batch to ``cancelling`` and the
    running attempt must terminate the process and record ``cancelled``."""
    config, sf, lib = _env(
        tmp_path, immich_server, uploader=_uploader(tmp_path, 'echo started; sleep 30; exit 0')
    )
    _album(sf, lib)
    (batch,) = _approved(sf, config)
    service = _service(sf, config)
    outcome = {}
    def _run():
        outcome["batch"] = service.run(batch["id"])
    runner = threading.Thread(target=_run)
    started = time.monotonic()
    runner.start()
    while service.get(batch["id"])["state"] != BatchState.RUNNING:
        assert time.monotonic() - started < 30, "the attempt never started"
        time.sleep(0.02)
    service.cancel(batch["id"])
    runner.join(timeout=30)
    assert not runner.is_alive(), "cancellation must not wait for the uploader's own timeout"
    assert outcome["batch"]["state"] == BatchState.CANCELLED
    assert all(item["state"] == ItemState.PENDING for item in outcome["batch"]["items"])
 def test_cancelling_a_planned_batch_never_starts_the_uploader(tmp_path, immich_server):
    config, sf, lib = _env(tmp_path, immich_server)
    _album(sf, lib)
    (batch,) = _approved(sf, config)
    service = _service(sf, config)
    cancelled = service.cancel(batch["id"])
    assert cancelled["state"] == BatchState.CANCELLED
    assert _argv(tmp_path) == []
 def test_a_cancelled_batch_can_be_run_again(tmp_path, immich_server):
    config, sf, lib = _env(tmp_path, immich_server)
    _album(sf, lib)
    (batch,) = _approved(sf, config)
    service = _service(sf, config)
    service.cancel(batch["id"])
    result = service.run(batch["id"])
    assert result["state"] == BatchState.SUCCEEDED and result["attempt_count"] == 1
 # ── restart / recovery ───────────────────────────────────────────────────────
 def test_an_interrupted_attempt_becomes_uncertain_not_failed(tmp_path, immich_server):
    """The process died mid-upload: Immich may hold the files, so the batch requires
    verification (US05-04) instead of a blind retry, and the lane is released."""
    config, sf, lib = _env(tmp_path, immich_server)
    _album(sf, lib)
    (batch,) = _approved(sf, config)
    with sf() as session:  # what a killed worker leaves behind
        session.get(UploadBatch, batch["id"]).state = BatchState.RUNNING
        session.commit()
    service = _service(sf, config)
    assert service.recover() == {"interrupted": 1}
    recovered = service.get(batch["id"])
    assert recovered["state"] == BatchState.UNKNOWN
    assert recovered["error_code"] == "interrupted"
    with pytest.raises(BatchError) as error:
        service.run(batch["id"])
    assert error.value.code == "requires_verification"
    assert _argv(tmp_path) == []
 def test_recovery_leaves_a_planned_batch_runnable(tmp_path, immich_server):
    config, sf, lib = _env(tmp_path, immich_server)
    _album(sf, lib)
    (batch,) = _approved(sf, config)
    service = _service(sf, config)
    assert service.recover() == {"interrupted": 0}
    assert service.run(batch["id"])["state"] == BatchState.SUCCEEDED
 def test_application_startup_recovers_an_interrupted_batch(tmp_path, immich_server):
    config, sf, lib = _env(tmp_path, immich_server)
    _album(sf, lib)
    (batch,) = _approved(sf, config)
    with sf() as session:
        session.get(UploadBatch, batch["id"]).state = BatchState.RUNNING
        session.commit()
    with TestClient(create_app(config)) as client:
        state = client.get(f"/api/v1/upload-batches/{batch['id']}").json()["state"]
    assert state == BatchState.UNKNOWN
 def test_a_killed_uploader_leaves_an_uncertain_batch(tmp_path, immich_server):
    """A real SIGKILL of the uploader process, not a simulated state write."""
    config, sf, lib = _env(
        tmp_path,
        immich_server,
        uploader=_uploader(tmp_path, 'echo "pid $$"; sleep 30; exit 0'),
    )
    _album(sf, lib)
    (batch,) = _approved(sf, config)
    service = _service(sf, config)
    outcome = {}
    runner = threading.Thread(target=lambda: outcome.update(batch=service.run(batch["id"])))
    runner.start()
    report = tmp_path / "data" / "uploads"
    deadline = time.monotonic() + 30
    pid = None
    while pid is None:
        assert time.monotonic() < deadline, "the uploader never announced itself"
        for log in report.glob("*.log"):
            text = log.read_text()
            if text.startswith("pid "):
                pid = int(text.split()[1])
        time.sleep(0.02)
    os.kill(pid, signal.SIGKILL)
    runner.join(timeout=30)
    # The process is gone with a non-zero status and no parsed report: the attempt
    # failed locally, and a restart classifies it honestly.
    assert outcome["batch"]["state"] in (BatchState.FAILED, BatchState.UNKNOWN)
    assert outcome["batch"]["exit_code"] != 0
 # ── worker integration ───────────────────────────────────────────────────────
 def test_the_worker_runs_a_queued_batch_on_the_upload_lane(tmp_path, immich_server):
    config, sf, lib = _env(tmp_path, immich_server)
    _album(sf, lib)
    (batch,) = _approved(sf, config)
    with TestClient(create_app(config)) as client:
        response = client.post(f"/api/v1/upload-batches/{batch['id']}/start")
    job_id = response.json()["job"]["id"]
    Worker(sf, worker_id="uploader-1", job_types=[UPLOAD_BATCH], config=config).run_once()
    assert JobService(sf).get(job_id)["state"] == JobState.SUCCEEDED
    assert _service(sf, config).get(batch["id"])["state"] == BatchState.SUCCEEDED
    assert len(_argv(tmp_path)) == 1
 # ── API surface ──────────────────────────────────────────────────────────────
 def test_api_creates_lists_and_reads_batches_without_secrets(tmp_path, immich_server):
    config, sf, lib = _env(tmp_path, immich_server)
    _album(sf, lib)
    with TestClient(create_app(config)) as client:
        token = client.post("/api/v1/upload-preflight", json={}).json()["token"]
        created = client.post("/api/v1/upload-batches", json={"token": token})
        listed = client.get("/api/v1/upload-batches")
        batch_id = created.json()["batches"][0]["id"]
        fetched = client.get(f"/api/v1/upload-batches/{batch_id}")
    assert created.status_code == 201
    assert listed.json()["batches"][0]["id"] == batch_id
    assert fetched.json()["state"] == BatchState.PLANNED
    assert SENTINEL_KEY not in created.text + listed.text + fetched.text
 def test_api_rejects_a_stale_token_and_an_unknown_batch(tmp_path, immich_server):
    config, sf, lib = _env(tmp_path, immich_server)
    _album(sf, lib)
    with TestClient(create_app(config)) as client:
        stale = client.post("/api/v1/upload-batches", json={"token": "v1:not-the-token"})
        missing = client.get("/api/v1/upload-batches/does-not-exist")
    assert stale.status_code == 409 and stale.json()["error"]["code"] == "stale_preflight"
    assert missing.status_code == 404
--- a/tests/integration/test_upload_preflight.py
+++ b/tests/integration/test_upload_preflight.py
@@ -1,511 +0,0 @@
 """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"
--- a/tests/story_traceability.json
+++ b/tests/story_traceability.json
@@ -103,12 +103,6 @@
    ],
    "US04-06": [
      "tests/e2e/test_phase_d_pipeline.py"
    ],
    "US05-01": [
      "tests/integration/test_upload_preflight.py"
    ],
    "US05-02": [
      "tests/integration/test_upload_batches.py"
    ]
  }
 }