US05-02: Orchestrate Album Upload Batches

migrations/versions/0008_upload_batches.py

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

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).run_forever()
+        Worker(create_session_factory(engine), worker_id=args.id, config=config).run_forever()
         return 0
 
     import uvicorn

photo_pipeline/api/app.py

@@ -34,6 +34,7 @@ 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"
 
@@ -50,6 +51,9 @@ 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:

photo_pipeline/api/routes/uploads.py

@@ -1,8 +1,9 @@
-"""Upload preflight API (US05-01).
+"""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. It uploads nothing — starting a batch is
-US05-02, so there is deliberately no start endpoint here.
+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
@@ -11,6 +12,13 @@ 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"])
@@ -23,17 +31,83 @@ class PreflightRequest(BaseModel):
     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 JSONResponse(
-            status_code=422,
-            content={"error": {"code": "unknown_album", "message": str(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))

photo_pipeline/integrations/immich_go.py

@@ -9,20 +9,42 @@ 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:
@@ -92,6 +114,103 @@ def redact(command: list[str]) -> list[str]:
     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]:

photo_pipeline/jobs/domain_handlers.py

@@ -1,10 +1,10 @@
-"""Domain job handlers: safety scoring and content analysis (US02-06).
+"""Domain job handlers: safety scoring, content analysis, uploads (US02-06, US05-02).
 
-Importing this module registers the ``safety_score`` and ``analysis`` job types so
-the generic worker can run them per item (one item = one ``asset_id``). Each handler
+Importing this module registers the ``safety_score``, ``analysis``, and
+``upload_batch`` job types so the generic worker can run them per item. 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.
+attempt, and an upload batch refuses to re-run an attempt whose outcome is unknown.
 
 Providers/models are the service defaults here (real NsfwModel / vision provider);
 tests exercise the services directly with injected fakes rather than the worker.
@@ -12,12 +12,15 @@ tests exercise the services directly with injected fakes rather than the worker.
 
 from __future__ import annotations
 
-from photo_pipeline.jobs.handlers import JobContext, register
+from photo_pipeline.jobs.handlers import Cancelled, 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:
@@ -32,5 +35,22 @@ 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)

photo_pipeline/jobs/handlers.py

@@ -34,6 +34,9 @@ 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

photo_pipeline/jobs/worker.py

@@ -30,6 +30,7 @@ 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)
@@ -37,6 +38,7 @@ 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."""
@@ -51,7 +53,9 @@ class Worker:
 
     def _process(self, job_id: str, job_type: str, token: int) -> None:
         handler = self.handlers[job_type]
-        ctx = JobContext(job_id, self.worker_id, token, self.service, self._session_factory)
+        ctx = JobContext(
+            job_id, self.worker_id, token, self.service, self._session_factory, self.config
+        )
         self._reset_interrupted_items(job_id, token)
 
         cancelled = False

photo_pipeline/models/__init__.py

@@ -14,6 +14,7 @@ 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__ = [
@@ -29,6 +30,8 @@ __all__ = [
     "RenamePlan",
     "RenameOperation",
     "Thumbnail",
+    "UploadBatch",
+    "UploadItem",
     "SafetyReview",
     "AnalysisResult",
 ]

photo_pipeline/models/uploads.py

@@ -0,0 +1,90 @@
+"""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()
+    )

photo_pipeline/services/hashing.py

@@ -24,7 +24,16 @@ _CHUNK = 1 << 20
 
 
 def sha256_file(path: Path | str) -> str:
-    digest = hashlib.sha256()
+    return _digest_file(path, 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)

photo_pipeline/services/upload_batches.py

@@ -0,0 +1,424 @@
+"""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
+        ],
+    }

tests/integration/test_upload_batches.py

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

tests/story_traceability.json

@@ -106,6 +106,9 @@
     ],
     "US05-01": [
       "tests/integration/test_upload_preflight.py"
+    ],
+    "US05-02": [
+      "tests/integration/test_upload_batches.py"
     ]
   }
 }