US04-05: Operate Rename Plans in the Browser

README.md

@@ -102,37 +102,3 @@ work_item/scripts/python -m pytest tests/e2e -m phase_c -q
 The deterministic naming provider is enabled only by test configuration
 (`PHOTO_PIPELINE_FAKE_NAMING_LOG`); without it the application falls back to the
 offline naming-policy name. Phase A and B suites remain green in the full run above.
-
-### Phase D acceptance gate
-
-Phase D (Epic E04: guarded renaming) is the first phase that changes the library on
-disk, so its gate is the strictest. One command runs the rename API journeys, the
-filesystem fault injection, and the browser suite:
-
-```bash
-work_item/scripts/python -m pytest tests/e2e -m phase_d -q
-```
-
-- `tests/e2e/test_phase_d_pipeline.py` drives a real server over HTTP: plan and
-  export, confirmation with the plan version and checksum (a stale token is refused
-  without touching disk), a valid apply, the case-only rename procedure, a collision
-  whose occupant survives, a source that changed after planning, and durability
-  across a full restart.
-- **Fault injection is real.** `PHOTO_PIPELINE_FAULT_AFTER=<journal state>` kills the
-  server process the instant that state is persisted. The suite crashes it at every
-  journal transition in turn (`moving`, `moved`, `database_updated`, `verified`),
-  starts a fresh process against the same database and library, and requires recovery
-  to converge from journal and disk evidence alone — with the asset set, the stable
-  IDs, and every content hash unchanged. Ambiguous evidence is never guessed: it stays
-  classified `manual` and keeps blocking. An unresolved rename is the cancellation
-  boundary — there is no cancel once a run starts, and unrelated mutations (album
-  proposal generation and approval) are refused with 409 `rename_recovery_required`
-  until it is resolved, while reads stay available.
-- `tests/e2e/test_renames_ui.py` covers the browser journeys: preview of every
-  affected path, confirmation carrying the server-issued token, apply with progress
-  and terminal verification, stale confirmation, collision, interruption, recovery,
-  rollback, keyboard confirmation, and the view still matching the journal after a
-  server restart.
-
-The fault barrier is test-only configuration; without `PHOTO_PIPELINE_FAULT_AFTER`
-the apply path has no crash points. Phases A–C remain green in the full run above.

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")

migrations/versions/0009_upload_report_outcomes.py

@@ -1,52 +0,0 @@
-"""Parsed uploader outcomes (US05-03).
-
-Revision ID: 0009_upload_report_outcomes
-Revises: 0008_upload_batches
-Create Date: 2026-08-16
-
-Per-item outcomes parsed from the immich-go report plus the parser evidence that
-produced them, so "what did Immich do with this file?" survives a restart.
-"""
-
-import sqlalchemy as sa
-from alembic import op
-
-revision = "0009_upload_report_outcomes"
-down_revision = "0008_upload_batches"
-branch_labels = None
-depends_on = None
-
-
-def upgrade() -> None:
-    # NULL parser = the uploader version has no pinned grammar; the batch is then
-    # requires_verification however the process exited.
-    op.add_column("upload_batches", sa.Column("parser", sa.String(), nullable=True))
-    op.add_column("upload_batches", sa.Column("parser_version", sa.Integer(), nullable=True))
-    op.add_column(
-        "upload_batches", sa.Column("parsed_at", sa.DateTime(timezone=True), nullable=True)
-    )
-    # verified | requires_verification
-    op.add_column("upload_batches", sa.Column("outcome_state", sa.String(), nullable=True))
-    op.add_column("upload_batches", sa.Column("outcome_counts", sa.String(), nullable=True))
-    op.add_column("upload_batches", sa.Column("report_counts", sa.String(), nullable=True))
-
-    # uploaded | upgraded | duplicate | skipped | failed | unknown
-    op.add_column("upload_items", sa.Column("outcome", sa.String(), nullable=True))
-    op.add_column("upload_items", sa.Column("evidence", sa.String(), nullable=True))
-    op.add_column(
-        "upload_items", sa.Column("outcome_at", sa.DateTime(timezone=True), nullable=True)
-    )
-
-
-def downgrade() -> None:
-    for column in ("outcome_at", "evidence", "outcome"):
-        op.drop_column("upload_items", column)
-    for column in (
-        "report_counts",
-        "outcome_counts",
-        "outcome_state",
-        "parsed_at",
-        "parser_version",
-        "parser",
-    ):
-        op.drop_column("upload_batches", column)

migrations/versions/0010_upload_verification.py

@@ -1,70 +0,0 @@
-"""Upload verification and manual resolution (US05-04).
-
-Revision ID: 0010_upload_verification
-Revises: 0009_upload_report_outcomes
-Create Date: 2026-08-16
-
-Per-item server evidence plus the append-only history that produced it, so an
-uncertain upload can be resolved without ever guessing that it succeeded.
-"""
-
-import sqlalchemy as sa
-from alembic import op
-
-revision = "0010_upload_verification"
-down_revision = "0009_upload_report_outcomes"
-branch_labels = None
-depends_on = None
-
-
-def upgrade() -> None:
-    op.add_column(
-        "upload_batches", sa.Column("verified_at", sa.DateTime(timezone=True), nullable=True)
-    )
-    # At least one uploaded file was edited afterwards: a visible warning that also
-    # blocks re-running the batch.
-    op.add_column(
-        "upload_batches",
-        sa.Column("stale_bytes", sa.Boolean(), nullable=False, server_default=sa.false()),
-    )
-
-    # present | absent | inconclusive | manual
-    op.add_column("upload_items", sa.Column("verification", sa.String(), nullable=True))
-    op.add_column(
-        "upload_items", sa.Column("verified_at", sa.DateTime(timezone=True), nullable=True)
-    )
-    op.add_column("upload_items", sa.Column("observed_sha256", sa.String(), nullable=True))
-    op.add_column(
-        "upload_items",
-        sa.Column("changed_after_upload", sa.Boolean(), nullable=False, server_default=sa.false()),
-    )
-
-    op.create_table(
-        "upload_verifications",
-        sa.Column("id", sa.String(), primary_key=True),
-        sa.Column(
-            "batch_id",
-            sa.String(),
-            sa.ForeignKey("upload_batches.id", ondelete="CASCADE"),
-            nullable=False,
-            index=True,
-        ),
-        sa.Column("asset_id", sa.String(), nullable=False),
-        sa.Column("action", sa.String(), nullable=False),
-        sa.Column("source", sa.String(), nullable=False),
-        sa.Column("result", sa.String(), nullable=False),
-        sa.Column("outcome", sa.String(), nullable=True),
-        sa.Column("evidence", sa.String(), nullable=False),
-        sa.Column("actor", sa.String(), nullable=True),
-        sa.Column(
-            "created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()
-        ),
-    )
-
-
-def downgrade() -> None:
-    op.drop_table("upload_verifications")
-    for column in ("changed_after_upload", "observed_sha256", "verified_at", "verification"):
-        op.drop_column("upload_items", column)
-    for column in ("stale_bytes", "verified_at"):
-        op.drop_column("upload_batches", column)

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

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")

photo_pipeline/api/routes/uploads.py

@@ -1,176 +0,0 @@
-"""Upload preflight, batch, and verification API (US05-01, US05-02, US05-04).
-
-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.
-
-``verify`` and ``resolve`` are the way out of an uncertain outcome; ``start``
-refuses one with ``409`` rather than letting the browser retry it.
-"""
-
-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.upload_verification import (
-    UploadVerificationService,
-    VerificationError,
-    retry_blockers,
-)
-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
-
-
-class ResolveRequest(BaseModel):
-    asset_id: str
-    # uploaded | upgraded | duplicate | skipped | failed | unknown
-    outcome: str
-    # What the operator actually checked, and who they are — both mandatory so a
-    # manual resolution can never look like server evidence.
-    evidence: str
-    actor: 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 _verification(request: Request) -> UploadVerificationService:
-    return UploadVerificationService(
-        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}")
-    # An uncertain outcome or bytes changed after upload must be resolved first
-    # (US05-04); the worker refuses them too, but the user is told here.
-    blocked = retry_blockers(batch)
-    if blocked:
-        return _error(409, blocked[0]["code"], blocked[0]["message"])
-    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}/verify")
-def verify_batch(batch_id: str, request: Request):
-    """Check the batch against Immich and the bytes on disk (US05-04)."""
-    try:
-        return _verification(request).verify(batch_id)
-    except VerificationError as error:
-        return _error(404 if error.code == "not_found" else 422, error.code, str(error))
-
-
-@router.post("/upload-batches/{batch_id}/resolve")
-def resolve_item(batch_id: str, body: ResolveRequest, request: Request):
-    """Record an operator's own verification of one item. Evidence is mandatory."""
-    try:
-        return _verification(request).resolve(
-            batch_id,
-            body.asset_id,
-            outcome=body.outcome,
-            evidence=body.evidence,
-            actor=body.actor,
-        )
-    except VerificationError as error:
-        return _error(404 if error.code == "not_found" else 422, error.code, str(error))
-
-
-@router.get("/upload-batches/{batch_id}/verifications")
-def list_verifications(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 {"verifications": _verification(request).history(batch_id)}
-
-
-@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/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:

photo_pipeline/integrations/immich_go.py

@@ -1,295 +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:`bulk_upload_check`
-uses the same client to ask Immich which uploaded bytes it already holds, which is
-the authoritative evidence behind upload verification (US05-04).
-
-: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
-# Immich's own deduplication endpoint: the authoritative answer to "do you already
-# have these exact bytes?" used to verify uncertain uploads (US05-04).
-BULK_CHECK_PATH = "/api/assets/bulk-upload-check"
-CHECK_TIMEOUT_SECONDS = 30.0
-CHECK_BATCH_SIZE = 500
-# 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 bulk_upload_check(
-    server_url: str,
-    api_key: str | None,
-    checksums: dict[str, str],
-    *,
-    timeout: float = CHECK_TIMEOUT_SECONDS,
-) -> dict:
-    """Ask Immich which of these exact bytes it already holds (US05-04).
-
-    ``checksums`` maps an application key (the asset id) to the SHA-1 of the bytes
-    that were uploaded — the digest Immich itself deduplicates on. The answer is
-    ``{"reachable", "detail", "present"}`` where ``present`` maps each key to
-    ``True`` (the server rejected it as a duplicate, so it holds those bytes),
-    ``False`` (the server would accept it, so it does not), or ``None`` (the server
-    answered something this adapter will not interpret).
-
-    An unreachable or unparsable server is reported, never guessed at: the caller
-    must treat it as uncertainty rather than absence.
-    """
-    if not server_url or not api_key:
-        return {"reachable": False, "detail": "no Immich credentials configured", "present": {}}
-    keys = list(checksums)
-    present: dict[str, bool | None] = {}
-    for start in range(0, len(keys), CHECK_BATCH_SIZE):
-        # ponytail: fixed chunk size; make it configurable if a server ever rejects it.
-        chunk = keys[start : start + CHECK_BATCH_SIZE]
-        payload = {"assets": [{"id": key, "checksum": checksums[key]} for key in chunk]}
-        request = urllib.request.Request(  # noqa: S310 — http(s) URL from configuration
-            server_url.rstrip("/") + BULK_CHECK_PATH,
-            data=json.dumps(payload).encode("utf-8"),
-            headers={"Content-Type": "application/json", "x-api-key": api_key},
-            method="POST",
-        )
-        try:
-            with urllib.request.urlopen(request, timeout=timeout) as response:  # noqa: S310
-                body = json.loads(response.read().decode("utf-8") or "{}")
-        except (urllib.error.URLError, OSError, ValueError, TimeoutError) as error:
-            return {"reachable": False, "detail": f"{type(error).__name__}: {error}", "present": {}}
-        results = body.get("results")
-        if not isinstance(results, list):
-            return {
-                "reachable": False,
-                "detail": "unrecognised bulk-upload-check response",
-                "present": {},
-            }
-        for result in results:
-            if not isinstance(result, dict) or result.get("id") not in checksums:
-                continue
-            present[result["id"]] = _holds_bytes(result)
-    return {"reachable": True, "detail": None, "present": present}
-
-
-def _holds_bytes(result: dict) -> bool | None:
-    """Whether one bulk-upload-check result means the server already has the file."""
-    action, reason = result.get("action"), result.get("reason")
-    if action == "reject":
-        # Only a duplicate proves possession; "unsupported-format" and friends say
-        # nothing about whether the bytes are there.
-        return True if reason == "duplicate" else None
-    return False if action == "accept" else None
-
-
-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,
-        )
-    )

photo_pipeline/integrations/immich_go_report.py

@@ -1,204 +0,0 @@
-"""immich-go report parsing (US05-03).
-
-The uploader's output is the only local evidence of what Immich did with each file,
-and its wording changes between releases (concept §15 "external integration
-risks"). So parsing is deliberately conservative:
-
-- **the parser is chosen by uploader version, not by guessing the format.** Each
-  supported version family maps to one adapter with a pinned line grammar. An
-  unrecognised version yields no adapter at all, which makes every item uncertain
-  rather than optimistically successful.
-- **an unmatched line is never success.** Lines the adapter does not recognise are
-  counted (``unparsed``) and kept as evidence; they never classify an item.
-- **evidence is bounded.** Each entry keeps a trimmed copy of the line that
-  classified it, and the number of entries is capped, so a looping uploader cannot
-  turn the report into unbounded database rows.
-
-Two grammars are supported. ``text-v1`` is the default human-readable log of the
-0.21/0.22 line, ``json-v1`` the structured log (``--log-type=json``) of 0.23/0.24.
-Both classify through one shared phrase table, so the vocabulary cannot drift
-between them:
-
-```text
-text-v1   INFO  uploaded /lib/rome/a.jpg
-          INFO  server has the same file /lib/rome/b.jpg
-          ERROR error uploading /lib/rome/e.jpg: connection reset
-          Uploaded 1, duplicates 1, errors 1
-
-json-v1   {"level":"INFO","msg":"uploaded","file":"/lib/rome/a.jpg"}
-          {"level":"INFO","msg":"report","counts":{"uploaded":1}}
-```
-
-Nothing here touches the database; :mod:`photo_pipeline.services.upload_reports`
-turns a parse result into durable per-item outcomes.
-"""
-
-from __future__ import annotations
-
-import json
-import re
-
-PARSER_VERSION = 1
-
-# Per-item outcomes. ``UNKNOWN`` is the safe default everywhere: it means the app
-# does not know what happened to that file and US05-04 must verify it.
-UPLOADED = "uploaded"
-UPGRADED = "upgraded"
-DUPLICATE = "duplicate"
-SKIPPED = "skipped"
-FAILED = "failed"
-UNKNOWN = "unknown"
-
-OUTCOMES = (UPLOADED, UPGRADED, DUPLICATE, SKIPPED, FAILED, UNKNOWN)
-
-# Longest/most specific phrases first: "server has an older file" also contains
-# "server has", and an error line about uploading also contains "upload".
-_PHRASES: tuple[tuple[str, str], ...] = (
-    ("server has the same file", DUPLICATE),
-    ("server has an older file", UPGRADED),
-    ("upgraded", UPGRADED),
-    ("duplicate", DUPLICATE),
-    ("discarded", SKIPPED),
-    ("skipped", SKIPPED),
-    ("error", FAILED),
-    ("failed", FAILED),
-    ("uploaded", UPLOADED),
-    ("upload", UPLOADED),
-)
-
-# Summary line of the text grammar: "Uploaded 3, duplicates 1, errors 2".
-_SUMMARY_WORDS = {
-    "uploaded": UPLOADED,
-    "upgraded": UPGRADED,
-    "duplicates": DUPLICATE,
-    "duplicate": DUPLICATE,
-    "skipped": SKIPPED,
-    "discarded": SKIPPED,
-    "errors": FAILED,
-    "error": FAILED,
-}
-_SUMMARY_PAIR = re.compile(r"([A-Za-z]+)\s+(\d+)")
-# A path is anything that looks absolute, up to an explanatory ": reason" tail.
-_PATH = re.compile(r"(/[^\s:][^:]*?)(?=:|$)")
-
-MAX_EVIDENCE_CHARS = 300
-# The report file is already byte-capped; this caps the rows it can produce.
-MAX_ENTRIES = 20_000
-
-
-def parser_for(uploader_version: str | None) -> str | None:
-    """Adapter name for a recorded uploader version, or ``None`` when unsupported.
-
-    Unsupported is not an error — it is the honest answer that this build's output
-    format was never pinned, and it makes the whole report uncertain.
-    """
-    match = re.search(r"(\d+)\.(\d+)", uploader_version or "")
-    if match is None:
-        return None
-    return _SUPPORTED.get(f"{match.group(1)}.{match.group(2)}")
-
-
-def parse(report: str, uploader_version: str | None) -> dict:
-    """Classify a raw report.
-
-    Returns ``{"parser", "parser_version", "supported", "entries", "counts",
-    "unparsed", "entries_truncated"}`` where ``entries`` is a list of
-    ``{"path", "outcome", "evidence"}`` and ``counts`` is the uploader's own
-    summary when it printed one (``None`` otherwise, never invented).
-    """
-    parser = parser_for(uploader_version)
-    result = {
-        "parser": parser,
-        "parser_version": PARSER_VERSION,
-        "supported": parser is not None,
-        "entries": [],
-        "counts": None,
-        "unparsed": 0,
-        "entries_truncated": False,
-    }
-    if parser is None:
-        return result
-    read_line = _text_line if parser == "text-v1" else _json_line
-    for raw in report.splitlines():
-        line = raw.strip()
-        if not line:
-            continue
-        path, outcome, counts = read_line(line)
-        if counts is not None:
-            # Later summaries win: the uploader prints its totals once, at the end.
-            result["counts"] = counts
-            continue
-        if path is None or outcome is None:
-            result["unparsed"] += 1
-            continue
-        if len(result["entries"]) >= MAX_ENTRIES:
-            result["entries_truncated"] = True
-            continue
-        result["entries"].append(
-            {"path": path, "outcome": outcome, "evidence": line[:MAX_EVIDENCE_CHARS]}
-        )
-    return result
-
-
-def _classify(text: str) -> str | None:
-    lowered = text.lower()
-    for phrase, outcome in _PHRASES:
-        if phrase in lowered:
-            return outcome
-    return None
-
-
-def _summary(text: str) -> dict[str, int] | None:
-    """Totals from a summary line, or ``None`` when the line is not one."""
-    counts: dict[str, int] = {}
-    for word, number in _SUMMARY_PAIR.findall(text):
-        outcome = _SUMMARY_WORDS.get(word.lower())
-        if outcome is None:
-            return None  # an unknown noun means this is not the summary grammar
-        counts[outcome] = counts.get(outcome, 0) + int(number)
-    return counts or None
-
-
-def _text_line(line: str) -> tuple[str | None, str | None, dict | None]:
-    path_match = _PATH.search(line)
-    if path_match is None:
-        return None, None, _summary(line)
-    path = path_match.group(1).strip()
-    # Classify from the words around the path, never from the path itself: an
-    # album called "errors" must not turn an upload into a failure.
-    context = line.replace(path, " ")
-    return path, _classify(context), None
-
-
-def _json_line(line: str) -> tuple[str | None, str | None, dict | None]:
-    try:
-        record = json.loads(line)
-    except ValueError:
-        return None, None, None
-    if not isinstance(record, dict):
-        return None, None, None
-    counts = record.get("counts")
-    if isinstance(counts, dict):
-        totals = {
-            _SUMMARY_WORDS[key.lower()]: int(value)
-            for key, value in counts.items()
-            if key.lower() in _SUMMARY_WORDS and isinstance(value, int)
-        }
-        return None, None, totals or None
-    path = record.get("file")
-    message = record.get("msg")
-    if not isinstance(path, str) or not isinstance(message, str):
-        return None, None, None
-    return path, _classify(message), None
-
-
-# Version family → adapter. Pinning this is the point: a build outside the list is
-# uncertain by construction (concept §15 "pin and record supported immich-go
-# versions").
-_SUPPORTED = {
-    "0.21": "text-v1",
-    "0.22": "text-v1",
-    "0.23": "json-v1",
-    "0.24": "json-v1",
-}
-SUPPORTED_VERSIONS = tuple(sorted(_SUPPORTED))

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)

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

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

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, UploadVerification
 from photo_pipeline.models.workflow import AnalysisResult, SafetyReview
 
 __all__ = [
@@ -30,9 +29,6 @@ __all__ = [
     "RenamePlan",
     "RenameOperation",
     "Thumbnail",
-    "UploadBatch",
-    "UploadItem",
-    "UploadVerification",
     "SafetyReview",
     "AnalysisResult",
 ]

photo_pipeline/models/uploads.py

@@ -1,151 +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)
-
-    # Parsed report evidence (US05-03). ``parser`` is NULL when the uploader's
-    # version has no pinned grammar; ``outcome_state`` is then
-    # ``requires_verification`` regardless of how the process exited.
-    parser: Mapped[str | None] = mapped_column(String)
-    parser_version: Mapped[int | None] = mapped_column(Integer)
-    parsed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
-    # verified | requires_verification; NULL until a report has been parsed.
-    outcome_state: Mapped[str | None] = mapped_column(String)
-    outcome_counts: Mapped[str | None] = mapped_column(String)  # JSON, from the items
-    report_counts: Mapped[str | None] = mapped_column(String)  # JSON, uploader's own
-
-    # Verification (US05-04). ``stale_bytes`` means at least one uploaded file has
-    # been edited since: the batch carries a visible warning and cannot be re-run.
-    verified_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
-    stale_bytes: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
-
-    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 — what the batch *process* did with this item;
-    # ``sent`` only means the uploader exited successfully.
-    state: Mapped[str] = mapped_column(String, nullable=False, default="pending")
-    # What the uploader's report says happened (US05-03): uploaded | upgraded |
-    # duplicate | skipped | failed | unknown. NULL before the report is parsed;
-    # ``unknown`` whenever the report does not classify this file — never success.
-    outcome: Mapped[str | None] = mapped_column(String)
-    evidence: Mapped[str | None] = mapped_column(String)  # the bounded report line
-    outcome_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
-
-    # Verification against the server (US05-04): present | absent | inconclusive |
-    # manual. NULL until the item has been verified; ``inconclusive`` whenever the
-    # server could not answer — which is never treated as success.
-    verification: Mapped[str | None] = mapped_column(String)
-    verified_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
-    # The bytes on disk at verification time, and whether they still are the bytes
-    # this batch uploaded.
-    observed_sha256: Mapped[str | None] = mapped_column(String)
-    changed_after_upload: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
-
-    updated_at: Mapped[datetime] = mapped_column(
-        DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now()
-    )
-
-
-class UploadVerification(Base):
-    """Append-only evidence for every verification and manual resolution (US05-04).
-
-    The item row is a projection of the latest answer; this table is the history
-    that answers "who decided this, on what evidence, and when?". Rows are never
-    updated or deleted, so a manual resolution can always be told apart from
-    server evidence.
-    """
-
-    __tablename__ = "upload_verifications"
-
-    id: Mapped[str] = mapped_column(String, primary_key=True)
-    batch_id: Mapped[str] = mapped_column(
-        ForeignKey("upload_batches.id", ondelete="CASCADE"), nullable=False, index=True
-    )
-    asset_id: Mapped[str] = mapped_column(String, nullable=False)
-    action: Mapped[str] = mapped_column(String, nullable=False)  # verify | resolve
-    source: Mapped[str] = mapped_column(String, nullable=False)  # immich_api | operator
-    # present | absent | inconclusive for a verify; the recorded outcome for a resolve.
-    result: Mapped[str] = mapped_column(String, nullable=False)
-    outcome: Mapped[str | None] = mapped_column(String)
-    evidence: Mapped[str] = mapped_column(String, nullable=False)
-    actor: Mapped[str | None] = mapped_column(String)
-    created_at: Mapped[datetime] = mapped_column(
-        DateTime(timezone=True), nullable=False, server_default=func.now()
-    )

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)

photo_pipeline/services/upload_batches.py

@@ -1,451 +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. :func:`~photo_pipeline.services.upload_verification.retry_blockers`
-  (US05-04) is what decides whether an attempt may start at all.
-
-Per-asset upload *results* are not interpreted here: after the attempt ends the
-report is handed to :class:`~photo_pipeline.services.upload_reports.
-UploadReportService` (US05-03), which classifies every file. The batch ``state``
-stays an honest description of what the *process* did; ``outcome_state`` says
-whether the report's evidence is complete enough to trust.
-"""
-
-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.upload_reports import UploadReportService
-from photo_pipeline.services.upload_verification import retry_blockers
-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)
-        # Retry policy (US05-04): a safe failure may run again, an uncertain outcome
-        # or bytes edited after upload may not.
-        blocked = retry_blockers(batch)
-        if blocked:
-            raise BatchError(blocked[0]["code"], blocked[0]["message"])
-        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)
-        # The report is the only evidence of what happened per file, so it is read
-        # while it is fresh (US05-03). A parse failure must not lose the batch
-        # outcome that was just recorded; the items simply stay unknown.
-        try:
-            UploadReportService(self._session_factory).ingest(batch_id)
-        except OSError:
-            pass
-        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,
-        # Parsed report evidence (US05-03): what the uploader said per file, and
-        # whether that evidence is complete enough to be trusted.
-        "parser": row.parser,
-        "parser_version": row.parser_version,
-        "parsed_at": row.parsed_at.isoformat() if row.parsed_at else None,
-        "outcome_state": row.outcome_state,
-        "outcome_counts": json.loads(row.outcome_counts) if row.outcome_counts else None,
-        "report_counts": json.loads(row.report_counts) if row.report_counts else None,
-        # Verification evidence (US05-04). ``stale_bytes`` is the visible warning
-        # that an uploaded file has since been edited.
-        "verified_at": row.verified_at.isoformat() if row.verified_at else None,
-        "stale_bytes": row.stale_bytes,
-        "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,
-                "outcome": item.outcome,
-                "evidence": item.evidence,
-                "outcome_at": item.outcome_at.isoformat() if item.outcome_at else None,
-                "verification": item.verification,
-                "verified_at": item.verified_at.isoformat() if item.verified_at else None,
-                "observed_sha256": item.observed_sha256,
-                "changed_after_upload": item.changed_after_upload,
-            }
-            for item in items
-        ],
-    }

photo_pipeline/services/upload_reports.py

@@ -1,140 +0,0 @@
-"""UploadReportService — durable per-item upload outcomes (US05-03).
-
-US05-02 proves the uploader ran and kept a bounded, credential-free report. This
-service is the step that reads that report and answers the operator's actual
-question — *what happened to each photo?* — without ever guessing in the app's
-favour:
-
-- **an item is ``unknown`` until the report says otherwise.** A file the report
-  never mentions, a report from an unpinned uploader version, and a line the
-  grammar does not recognise all leave the item ``unknown``. Exit code 0 is
-  evidence the *process* ended well, never evidence that a particular file reached
-  Immich (concept §15 "wrong upload state").
-- **counts are reconciled, not trusted.** The uploader's own summary is stored
-  next to the totals derived from the items. A disagreement makes the batch
-  ``requires_verification`` even when every line parsed.
-- **reprocessing is idempotent.** Outcomes are keyed by ``(batch_id, asset_id)``
-  and rewritten in place, so parsing the same report again — after a restart, or
-  because the operator asked — converges on the same rows and the same counts.
-
-The batch's own ``state`` still describes the process (US05-02's contract);
-``outcome_state`` describes the evidence. Turning ``requires_verification`` into
-verification, retry, and manual resolution is US05-04.
-"""
-
-from __future__ import annotations
-
-import json
-import os
-from datetime import datetime, timezone
-from pathlib import Path
-
-from sqlalchemy import select
-from sqlalchemy.orm import sessionmaker
-
-from photo_pipeline.integrations import immich_go_report as report_parser
-from photo_pipeline.models import UploadBatch, UploadItem
-
-VERIFIED = "verified"
-REQUIRES_VERIFICATION = "requires_verification"
-
-
-def _now() -> datetime:
-    return datetime.now(timezone.utc)
-
-
-def _key(path: str) -> str:
-    return os.path.normpath(path)
-
-
-class UploadReportService:
-    def __init__(self, session_factory: sessionmaker) -> None:
-        self._session_factory = session_factory
-
-    def ingest(self, batch_id: str) -> dict:
-        """Parse this batch's report and persist an outcome for every item.
-
-        Returns ``{"parser", "parser_version", "outcome_state", "counts",
-        "report_counts", "unmatched", "unparsed"}``. Safe to call repeatedly.
-        """
-        with self._session_factory() as session:
-            batch = session.get(UploadBatch, batch_id)
-            if batch is None:
-                raise KeyError(f"unknown upload batch {batch_id!r}")
-            items = list(session.scalars(select(UploadItem).where(UploadItem.batch_id == batch_id)))
-            parsed = report_parser.parse(_read_report(batch.report_path), batch.uploader_version)
-
-            by_path = {_key(entry["path"]): entry for entry in parsed["entries"]}
-            # Basenames are the fallback: an uploader may log a path relative to the
-            # folder it was given. Ambiguous basenames are dropped rather than
-            # guessed at.
-            by_name: dict[str, dict] = {}
-            for key, entry in by_path.items():
-                name = os.path.basename(key)
-                by_name[name] = None if name in by_name else entry
-
-            matched: set[str] = set()
-            counts = {outcome: 0 for outcome in report_parser.OUTCOMES}
-            for item in items:
-                key = _key(item.path)
-                entry = by_path.get(key) or by_name.get(os.path.basename(key))
-                if entry is not None:
-                    matched.add(_key(entry["path"]))
-                item.outcome = entry["outcome"] if entry else report_parser.UNKNOWN
-                item.evidence = entry["evidence"] if entry else None
-                item.outcome_at = _now()
-                counts[item.outcome] += 1
-
-            unmatched = sorted(set(by_path) - matched)
-            reconciled = _reconciles(parsed["counts"], counts)
-            state = (
-                VERIFIED
-                if (
-                    parsed["supported"]
-                    and items
-                    and not counts[report_parser.UNKNOWN]
-                    and not unmatched
-                    and not parsed["entries_truncated"]
-                    and reconciled
-                )
-                else REQUIRES_VERIFICATION
-            )
-
-            batch.parser = parsed["parser"]
-            batch.parser_version = parsed["parser_version"]
-            batch.parsed_at = _now()
-            batch.outcome_state = state
-            batch.outcome_counts = json.dumps(counts, sort_keys=True)
-            batch.report_counts = (
-                json.dumps(parsed["counts"], sort_keys=True) if parsed["counts"] else None
-            )
-            session.commit()
-
-            return {
-                "parser": parsed["parser"],
-                "parser_version": parsed["parser_version"],
-                "outcome_state": state,
-                "counts": counts,
-                "report_counts": parsed["counts"],
-                "unmatched": unmatched,
-                "unparsed": parsed["unparsed"],
-            }
-
-
-def _read_report(report_path: str | None) -> str:
-    """The raw report, or ``""`` when the attempt never produced one."""
-    if not report_path:
-        return ""
-    path = Path(report_path)
-    return path.read_text(errors="replace") if path.exists() else ""
-
-
-def _reconciles(report_counts: dict | None, derived: dict) -> bool:
-    """Whether the uploader's own totals agree with the per-item outcomes.
-
-    No summary is not a disagreement — most of the uncertainty it would catch is
-    already caught by unmatched entries and unknown items.
-    """
-    if not report_counts:
-        return True
-    return all(derived.get(outcome) == total for outcome, total in report_counts.items())

photo_pipeline/services/upload_verification.py

@@ -1,349 +0,0 @@
-"""UploadVerificationService — resolving uncertain uploads (US05-04).
-
-US05-03 leaves a batch honest but sometimes uncertain: a killed uploader, an
-unpinned report grammar, or a file the report never mentioned all end as
-``unknown``. Retrying such a batch is the dangerous move — Immich may already hold
-the files, and a blind retry is how a lost response turns into a second server
-asset. So this service resolves uncertainty *before* anything is re-run:
-
-- **evidence beats the report.** Verification asks Immich itself whether it holds
-  the exact SHA-1 the batch recorded before uploading. That answer is authoritative
-  over the uploader's text, in both directions: present makes an unknown item
-  ``uploaded``, absent makes it ``failed`` and therefore safe to retry.
-- **no answer is never "no".** An unreachable server, missing credentials, or a
-  response this adapter will not interpret leave the item ``inconclusive``. The
-  batch stays ``unknown_requires_verification`` and stays un-runnable.
-- **changed bytes are a stale warning, not a silent re-upload.** Verification
-  re-hashes what is on disk. A file edited after its upload is flagged, the batch
-  is marked ``stale_bytes``, and re-running it is refused: uploading again would
-  create or upgrade a server asset the user never approved (concept §8).
-- **manual resolution is evidence, not permission.** An operator may record what
-  they checked in Immich, but only with a non-empty note and their identity, and
-  every decision is appended to an immutable history alongside the server answers.
-
-Retry policy lives in :func:`retry_blockers`, which :class:`UploadBatchService`
-enforces before every attempt and the API surfaces as ``409``.
-"""
-
-from __future__ import annotations
-
-import uuid
-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.integrations import immich_go_report as report_parser
-from photo_pipeline.models import UploadBatch, UploadItem, UploadVerification
-from photo_pipeline.services.hashing import sha256_file
-from photo_pipeline.services.upload_reports import REQUIRES_VERIFICATION, VERIFIED
-
-PRESENT = "present"
-ABSENT = "absent"
-INCONCLUSIVE = "inconclusive"
-MANUAL = "manual"
-
-MAX_EVIDENCE_CHARS = 500
-
-
-def _now() -> datetime:
-    return datetime.now(timezone.utc)
-
-
-class VerificationError(RuntimeError):
-    """The request cannot be carried out (unknown batch/item, missing evidence)."""
-
-    def __init__(self, code: str, message: str) -> None:
-        super().__init__(message)
-        self.code = code
-
-
-class UploadVerificationService:
-    def __init__(self, session_factory: sessionmaker, *, config: Config) -> None:
-        self._session_factory = session_factory
-        self._config = config
-
-    # ── verification ──────────────────────────────────────────────────────────
-
-    def verify(self, batch_id: str) -> dict:
-        """Check every item of a batch against the server and the bytes on disk.
-
-        Returns ``{"batch_id", "state", "outcome_state", "stale_bytes",
-        "server_reachable", "detail", "counts", "items"}``. Safe to call
-        repeatedly: the same evidence produces the same rows, and each run appends
-        its own history entries.
-        """
-        with self._session_factory() as session:
-            batch = session.get(UploadBatch, batch_id)
-            if batch is None:
-                raise VerificationError("not_found", f"unknown upload batch {batch_id!r}")
-            items = list(
-                session.scalars(
-                    select(UploadItem)
-                    .where(UploadItem.batch_id == batch_id)
-                    .order_by(UploadItem.path)
-                )
-            )
-            answer = immich_go.bulk_upload_check(
-                self._config.immich_server_url,
-                self._config.immich_api_key.get_secret_value()
-                if self._config.immich_api_key
-                else None,
-                {item.asset_id: item.sha1 for item in items if item.sha1},
-            )
-            detail = answer["detail"]
-
-            for item in items:
-                observed, changed = _current_bytes(item)
-                held = answer["present"].get(item.asset_id)
-                if item.sha1 is None:
-                    result, evidence = INCONCLUSIVE, "no upload hash was recorded for this file"
-                elif held is True:
-                    result, evidence = PRESENT, f"Immich holds sha1 {item.sha1}"
-                elif held is False:
-                    result, evidence = ABSENT, f"Immich does not hold sha1 {item.sha1}"
-                else:
-                    result = INCONCLUSIVE
-                    evidence = detail or "the server did not classify these bytes"
-
-                item.verification = result
-                item.verified_at = _now()
-                item.observed_sha256 = observed
-                item.changed_after_upload = changed
-                # The server is authoritative over the report text — but only when
-                # it actually answered.
-                if result == PRESENT:
-                    item.outcome = report_parser.UPLOADED
-                    item.evidence = evidence
-                    item.outcome_at = _now()
-                elif result == ABSENT:
-                    item.outcome = report_parser.FAILED
-                    item.evidence = evidence
-                    item.outcome_at = _now()
-                session.add(
-                    _event(
-                        batch_id,
-                        item.asset_id,
-                        action="verify",
-                        source="immich_api",
-                        result=result,
-                        outcome=item.outcome,
-                        evidence=evidence,
-                    )
-                )
-
-            _resolve_batch(batch, items)
-            session.commit()
-            return _report(batch, items, server_reachable=answer["reachable"], detail=detail)
-
-    # ── manual resolution ─────────────────────────────────────────────────────
-
-    def resolve(
-        self, batch_id: str, asset_id: str, *, outcome: str, evidence: str, actor: str
-    ) -> dict:
-        """Record an operator's own verification of one item.
-
-        ``evidence`` and ``actor`` are mandatory: a manual resolution is only worth
-        keeping if it says what was checked and who checked it.
-        """
-        if outcome not in report_parser.OUTCOMES:
-            raise VerificationError("invalid_outcome", f"unknown upload outcome {outcome!r}")
-        evidence = (evidence or "").strip()
-        actor = (actor or "").strip()
-        if not evidence:
-            raise VerificationError("evidence_required", "a manual resolution must record evidence")
-        if not actor:
-            raise VerificationError("actor_required", "a manual resolution must record its author")
-
-        with self._session_factory() as session:
-            batch = session.get(UploadBatch, batch_id)
-            if batch is None:
-                raise VerificationError("not_found", f"unknown upload batch {batch_id!r}")
-            item = session.get(UploadItem, {"batch_id": batch_id, "asset_id": asset_id})
-            if item is None:
-                raise VerificationError("not_found", f"{asset_id!r} is not part of this batch")
-
-            item.outcome = outcome
-            item.evidence = evidence[:MAX_EVIDENCE_CHARS]
-            item.outcome_at = _now()
-            item.verification = MANUAL
-            item.verified_at = _now()
-            session.add(
-                _event(
-                    batch_id,
-                    asset_id,
-                    action="resolve",
-                    source="operator",
-                    result=MANUAL,
-                    outcome=outcome,
-                    evidence=evidence,
-                    actor=actor,
-                )
-            )
-
-            items = list(
-                session.scalars(
-                    select(UploadItem)
-                    .where(UploadItem.batch_id == batch_id)
-                    .order_by(UploadItem.path)
-                )
-            )
-            _resolve_batch(batch, items)
-            session.commit()
-            return _report(batch, items, server_reachable=None, detail=None)
-
-    # ── history ───────────────────────────────────────────────────────────────
-
-    def history(self, batch_id: str) -> list[dict]:
-        """Every verification and resolution recorded for this batch, oldest first."""
-        with self._session_factory() as session:
-            rows = session.scalars(
-                select(UploadVerification)
-                .where(UploadVerification.batch_id == batch_id)
-                .order_by(UploadVerification.created_at, UploadVerification.id)
-            )
-            return [
-                {
-                    "id": row.id,
-                    "asset_id": row.asset_id,
-                    "action": row.action,
-                    "source": row.source,
-                    "result": row.result,
-                    "outcome": row.outcome,
-                    "evidence": row.evidence,
-                    "actor": row.actor,
-                    "created_at": row.created_at.isoformat() if row.created_at else None,
-                }
-                for row in rows
-            ]
-
-
-# ── retry policy ─────────────────────────────────────────────────────────────
-
-
-def retry_blockers(batch: dict) -> list[dict]:
-    """Why this batch may not be (re)run, in the order the user should fix them.
-
-    A plain uploader failure is a *safe* failure: nothing uncertain happened, so it
-    is retryable. An uncertain outcome and edited bytes are not.
-    """
-    from photo_pipeline.services.upload_batches import BatchState, RUNNABLE_STATES
-
-    blockers: list[dict] = []
-    if batch["state"] == BatchState.UNKNOWN:
-        blockers.append(
-            {
-                "code": "requires_verification",
-                "message": "this upload's outcome is uncertain; verify it before retrying",
-            }
-        )
-    if batch.get("stale_bytes"):
-        blockers.append(
-            {
-                "code": "changed_after_upload",
-                "message": "files in this batch changed after they were uploaded; "
-                "re-approve them through a fresh preflight",
-            }
-        )
-    if not blockers and batch["state"] not in RUNNABLE_STATES:
-        blockers.append({"code": "not_runnable", "message": f"batch is {batch['state']}"})
-    return blockers
-
-
-# ── internals ────────────────────────────────────────────────────────────────
-
-
-def _current_bytes(item: UploadItem) -> tuple[str | None, bool]:
-    """``(hash on disk now, changed since upload)``. A missing file counts as changed."""
-    path = Path(item.path)
-    if not path.exists():
-        return None, item.sha256 is not None
-    observed = sha256_file(path)
-    return observed, bool(item.sha256 and observed != item.sha256)
-
-
-def _event(
-    batch_id: str,
-    asset_id: str,
-    *,
-    action: str,
-    source: str,
-    result: str,
-    outcome: str | None,
-    evidence: str,
-    actor: str | None = None,
-) -> UploadVerification:
-    return UploadVerification(
-        id=str(uuid.uuid4()),
-        batch_id=batch_id,
-        asset_id=asset_id,
-        action=action,
-        source=source,
-        result=result,
-        outcome=outcome,
-        evidence=evidence[:MAX_EVIDENCE_CHARS],
-        actor=actor,
-        created_at=_now(),
-    )
-
-
-def _resolve_batch(batch: UploadBatch, items: list[UploadItem]) -> None:
-    """Fold the item evidence back into the batch's own state.
-
-    An uncertain batch only leaves that state once every item is accounted for:
-    all present makes it succeeded, any absent makes it a safe failure to retry,
-    and a single inconclusive item keeps it uncertain.
-    """
-    from photo_pipeline.services.upload_batches import BatchState
-
-    batch.stale_bytes = any(item.changed_after_upload for item in items)
-    batch.verified_at = _now()
-    unresolved = [item for item in items if item.outcome in (None, report_parser.UNKNOWN)]
-    batch.outcome_state = VERIFIED if items and not unresolved else REQUIRES_VERIFICATION
-
-    if batch.state != BatchState.UNKNOWN or unresolved:
-        return
-    if any(item.outcome == report_parser.FAILED for item in items):
-        batch.state = BatchState.FAILED
-        batch.error_code = "verified_incomplete"
-        batch.error_message = "verification proved some files never reached Immich"
-    else:
-        batch.state = BatchState.SUCCEEDED
-        batch.error_code = batch.error_message = None
-
-
-def _report(
-    batch: UploadBatch,
-    items: list[UploadItem],
-    *,
-    server_reachable: bool | None,
-    detail: str | None,
-) -> dict:
-    counts: dict[str, int] = {}
-    for item in items:
-        key = item.verification or "unverified"
-        counts[key] = counts.get(key, 0) + 1
-    return {
-        "batch_id": batch.id,
-        "state": batch.state,
-        "outcome_state": batch.outcome_state,
-        "stale_bytes": batch.stale_bytes,
-        "server_reachable": server_reachable,
-        "detail": detail,
-        "counts": counts,
-        "items": [
-            {
-                "asset_id": item.asset_id,
-                "path": item.path,
-                "verification": item.verification,
-                "outcome": item.outcome,
-                "evidence": item.evidence,
-                "changed_after_upload": item.changed_after_upload,
-                "verified_at": item.verified_at.isoformat() if item.verified_at else None,
-            }
-            for item in items
-        ],
-    }

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}"

tests/e2e/_pipeline_harness.py

@@ -14,9 +14,7 @@ import socket
 import subprocess
 import sys
 import time
-import uuid
 from dataclasses import dataclass
-from datetime import datetime, timezone
 from pathlib import Path
 
 import httpx
@@ -161,88 +159,3 @@ def wait_until(predicate, *, timeout: float = 20, interval: float = 0.1):
             return value
         time.sleep(interval)
     raise AssertionError("condition not met before timeout")
-
-
-# ── Phase D: an analysed album, ready to be named and renamed ────────────────
-
-NOW = datetime(2026, 1, 1, tzinfo=timezone.utc)
-
-
-class session_factory:
-    """Session factory against a seeded database, for the few things a test has to
-    set up or inspect below the API — journal states, mainly."""
-
-    def __init__(self, seeded: Seeded) -> None:
-        self._seeded = seeded
-
-    def __enter__(self):
-        from photo_pipeline.config import Config
-        from photo_pipeline.db import create_db_engine, create_session_factory, run_migrations
-
-        config = Config.from_env(
-            {
-                "PHOTO_PIPELINE_DATA_DIR": str(self._seeded.data),
-                "PHOTO_PIPELINE_LIBRARY_ROOTS": str(self._seeded.lib),
-            }
-        )
-        run_migrations(config.database_url)
-        self._engine = create_db_engine(config.database_url)
-        return create_session_factory(self._engine)
-
-    def __exit__(self, *_):
-        self._engine.dispose()
-        return False
-
-
-def seed_album(tmp_path: Path, album: str = "rome", names: tuple[str, ...] = ("a.jpg", "b.jpg")):
-    """A library holding one album folder whose photos are confirmed SFW and analysed
-    — the state a naming proposal, and therefore a rename plan, is built from."""
-    from sqlalchemy import select
-
-    from photo_pipeline.models import AnalysisResult, Asset, SafetyReview
-    from photo_pipeline.services.inventory import InventoryService
-
-    seeded = seed_library(tmp_path, {}, {})
-    folder = seeded.lib / album
-    folder.mkdir(parents=True)
-    for index, name in enumerate(names):
-        image(folder / name, index + 1)
-
-    with session_factory(seeded) as sf:
-        InventoryService(sf).scan(seeded.lib)
-        with sf() as session:
-            rows = list(session.execute(select(Asset.id, Asset.current_path)).all())
-            for asset_id, path in rows:
-                session.add(
-                    SafetyReview(
-                        id=str(uuid.uuid4()), asset_id=asset_id, decision="sfw", created_at=NOW
-                    )
-                )
-                session.add(
-                    AnalysisResult(
-                        asset_id=asset_id,
-                        status="analyzed",
-                        description=f"a view of {path}",
-                        tags='["ruins", "city"]',
-                        approx_year=2019,
-                        location_hint="Rome",
-                    )
-                )
-            session.commit()
-        seeded.asset_ids.update({Path(path).stem: aid for aid, path in rows})
-    return seeded
-
-
-def approve_album(base: str, *, album: str = "rome", name: str) -> None:
-    """Generate a proposal, set its final name, and approve it over HTTP."""
-    httpx.post(f"{base}/api/v1/albums/proposals", json={}, timeout=10).raise_for_status()
-    for payload, route in (
-        ({"name": name}, "edit"),
-        ({}, "approve"),
-    ):
-        current = httpx.get(f"{base}/api/v1/albums/proposals/{album}", timeout=10).json()
-        httpx.post(
-            f"{base}/api/v1/albums/proposals/{album}/{route}",
-            json={**payload, "expected_version": current["version"]},
-            timeout=10,
-        ).raise_for_status()

tests/e2e/test_phase_d_pipeline.py

@@ -1,377 +0,0 @@
-"""Phase D end-to-end acceptance (US04-06): guarded renaming, black box.
-
-Every journey here drives a real ``photo_pipeline serve`` child process over HTTP —
-plan, export, confirm, apply, collide, go stale, crash, recover, roll back. The
-crashes are real: the server is killed by the ``PHOTO_PIPELINE_FAULT_AFTER`` barrier
-at each persisted journal transition in turn, then a fresh process is started against
-the same database and library and has to reconcile the wreckage from evidence alone.
-
-Photos really move. After every journey the assertions read the filesystem and the
-inventory back: the asset set, the stable IDs, and the content hashes must be exactly
-what they were before, only at new paths.
-"""
-
-from __future__ import annotations
-
-import httpx
-import pytest
-
-from tests.e2e._pipeline_harness import Server, approve_album, seed_album, session_factory
-
-pytestmark = pytest.mark.phase_d
-
-TIMEOUT = 10
-APPROVED = "2019 Rome"
-CRASH_POINTS = ["moving", "moved", "database_updated", "verified"]
-
-
-@pytest.fixture
-def server(tmp_path):
-    seeded = seed_album(tmp_path)
-    running = Server(seeded).start()
-    running.seeded = seeded
-    try:
-        yield running
-    finally:
-        running.stop()
-
-
-# ── helpers ──────────────────────────────────────────────────────────────────
-
-
-def _plan(base) -> dict:
-    response = httpx.post(f"{base}/api/v1/rename-plans", timeout=TIMEOUT)
-    response.raise_for_status()
-    return response.json()
-
-
-def _get_plan(base, plan_id) -> dict:
-    return httpx.get(f"{base}/api/v1/rename-plans/{plan_id}", timeout=TIMEOUT).json()
-
-
-def _apply(base, plan, **body):
-    payload = {"expected_version": plan["version"], **body}
-    return httpx.post(
-        f"{base}/api/v1/rename-plans/{plan['id']}/apply", json=payload, timeout=TIMEOUT
-    )
-
-
-def _inventory(base) -> dict[str, dict]:
-    """Every asset by stable ID, so identity can be compared across a rename."""
-    items = httpx.get(
-        f"{base}/api/v1/inventory/assets", params={"limit": 200}, timeout=TIMEOUT
-    ).json()["items"]
-    return {item["id"]: item for item in items}
-
-
-def _content(root) -> dict[str, bytes]:
-    return {
-        str(path.relative_to(root)): path.read_bytes()
-        for path in sorted(root.rglob("*"))
-        if path.is_file()
-    }
-
-
-def _recovery(base) -> dict:
-    return httpx.get(f"{base}/api/v1/rename-recovery", timeout=TIMEOUT).json()
-
-
-def _journal_states(seeded, plan_id) -> list[str]:
-    from photo_pipeline.services.rename_journal import RenameJournal
-
-    with session_factory(seeded) as sf:
-        return [row["journal_state"] for row in RenameJournal(sf).operations(plan_id)]
-
-
-# ── US04-01: plan and export ─────────────────────────────────────────────────
-
-
-def test_plan_and_export_describe_every_move_without_touching_the_library(server):
-    before = _content(server.seeded.lib)
-    approve_album(server.base, name=APPROVED)
-    plan = _plan(server.base)
-
-    assert plan["state"] == "validated" and plan["operation_count"] == 1
-    operation = plan["operations"][0]
-    assert operation["source_path"].endswith("/rome")
-    assert operation["destination_path"].endswith(f"/{APPROVED}")
-    assert operation["asset_count"] == 2
-    assert set(operation["asset_ids"]) == set(_inventory(server.base))
-
-    export = httpx.get(
-        f"{server.base}/api/v1/rename-plans/{plan['id']}/export", timeout=TIMEOUT
-    ).json()
-    assert export["schema_version"] == 1
-    assert export["checksum"] == plan["checksum"]
-    assert [op["source_path"] for op in export["operations"]] == [operation["source_path"]]
-    # Portable evidence must not smuggle out anything sensitive.
-    assert "token" not in str(export).lower() and "key" not in str(export).lower()
-
-    # Planning is a preview: not one byte moved.
-    assert _content(server.seeded.lib) == before
-
-
-# ── US04-03: confirmation and apply ──────────────────────────────────────────
-
-
-def test_apply_requires_the_current_confirmation_token(server):
-    approve_album(server.base, name=APPROVED)
-    plan = _plan(server.base)
-
-    stale = _apply(server.base, {**plan, "version": plan["version"] + 7})
-    assert stale.status_code == 409 and stale.json()["error"]["code"] == "version_conflict"
-
-    wrong_checksum = _apply(server.base, plan, expected_checksum="0" * 64)
-    assert wrong_checksum.status_code == 409
-
-    assert (server.seeded.lib / "rome").is_dir(), "a refused confirmation moves nothing"
-
-
-def test_a_valid_apply_preserves_ids_hashes_and_the_asset_set(server):
-    approve_album(server.base, name=APPROVED)
-    before = _inventory(server.base)
-    before_content = _content(server.seeded.lib)
-    plan = _plan(server.base)
-
-    applied = _apply(server.base, plan, expected_checksum=plan["checksum"]).json()
-    assert applied["applied"] == 1 and applied["failed"] == 0 and applied["state"] == "applied"
-
-    after = _inventory(server.base)
-    assert set(after) == set(before), "renaming must not change asset identity"
-    assert {item["current_sha256"] for item in after.values()} == {
-        item["current_sha256"] for item in before.values()
-    }
-    assert all(APPROVED in item["current_path"] for item in after.values())
-    # Same bytes, new folder — nothing was rewritten in the move.
-    assert _content(server.seeded.lib) == {
-        key.replace("rome/", f"{APPROVED}/"): value for key, value in before_content.items()
-    }
-    assert _journal_states(server.seeded, plan["id"]) == ["complete"]
-
-
-def test_a_case_only_rename_applies_on_a_case_insensitive_filesystem(tmp_path):
-    seeded = seed_album(tmp_path, album="rome")
-    running = Server(seeded).start()
-    try:
-        approve_album(running.base, name="Rome")
-        plan = _plan(running.base)
-        assert plan["operations"][0]["case_only"] is True
-
-        assert _apply(running.base, plan).json()["applied"] == 1
-        entries = {path.name for path in seeded.lib.iterdir()}
-        assert "Rome" in entries
-        # The staged intermediate name must not survive the procedure.
-        assert not any(name.startswith(".rename-") for name in entries)
-        assert all("/Rome/" in item["current_path"] for item in _inventory(running.base).values())
-    finally:
-        running.stop()
-
-
-def test_a_collision_is_refused_and_the_occupant_survives(server):
-    approve_album(server.base, name=APPROVED)
-    occupied = server.seeded.lib / APPROVED
-    occupied.mkdir()
-    (occupied / "precious.jpg").write_bytes(b"do not lose me")
-
-    plan = _plan(server.base)
-    assert plan["state"] == "invalid" and "destination_exists" in plan["blockers"]
-
-    refused = _apply(server.base, plan)
-    assert refused.status_code == 422 and refused.json()["error"]["code"] == "cannot_apply"
-    assert (occupied / "precious.jpg").read_bytes() == b"do not lose me"
-    assert (server.seeded.lib / "rome").is_dir()
-
-
-def test_a_source_that_changed_after_planning_is_refused(server):
-    approve_album(server.base, name=APPROVED)
-    plan = _plan(server.base)
-    # The plan recorded per-asset hashes; the file changes before confirmation.
-    (server.seeded.lib / "rome" / "a.jpg").write_bytes(b"tampered")
-
-    result = _apply(server.base, plan).json()
-    assert result["failed"] == 1 and result["applied"] == 0
-    assert (server.seeded.lib / "rome").is_dir(), "a failed precondition leaves the source alone"
-    operation = _get_plan(server.base, plan["id"])["operations"][0]
-    assert operation["journal_state"] == "failed"
-    assert operation["error_code"] == "source_changed"
-
-
-# ── US04-04: crash points, recovery, rollback ────────────────────────────────
-
-
-def _crash_during_apply(seeded, plan, state):
-    """Apply with the fault barrier armed: the server dies at ``state``, mid-move."""
-    crashing = Server(seeded, extra_env={"PHOTO_PIPELINE_FAULT_AFTER": state}).start()
-    try:
-        with pytest.raises(httpx.HTTPError):
-            _apply(crashing.base, plan)
-    finally:
-        crashing.stop()
-    assert crashing.proc is None
-
-
-@pytest.mark.parametrize("crash_point", CRASH_POINTS)
-def test_every_journal_crash_point_recovers_without_losing_content(tmp_path, crash_point):
-    seeded = seed_album(tmp_path)
-    first = Server(seeded).start()
-    try:
-        approve_album(first.base, name=APPROVED)
-        before = _inventory(first.base)
-        before_bytes = sorted(_content(seeded.lib).values())
-        plan = _plan(first.base)
-    finally:
-        first.stop()
-
-    _crash_during_apply(seeded, plan, crash_point)
-
-    # A brand-new process, no in-memory state: everything comes from the journal.
-    restarted = Server(seeded).start()
-    try:
-        recovery = _recovery(restarted.base)
-        assert recovery["items"], f"a crash at {crash_point} must leave visible evidence"
-        assert recovery["items"][0]["journal_state"] == crash_point
-        # Only a crash that could have left the library half-renamed blocks other
-        # work. `verified` is past every filesystem and database change — the move
-        # is done and checked, just not flagged complete — so it blocks nothing.
-        assert recovery["blocks_mutation"] is (crash_point != "verified")
-
-        resolved = httpx.post(
-            f"{restarted.base}/api/v1/rename-recovery/resolve", timeout=TIMEOUT
-        ).json()
-        assert resolved["manual"] == 0, "an interrupted rename must be decidable from evidence"
-
-        # Recovery is idempotent: running it again changes nothing.
-        assert _recovery(restarted.base)["blocks_mutation"] is False
-        httpx.post(f"{restarted.base}/api/v1/rename-recovery/resolve", timeout=TIMEOUT)
-
-        # A resumable crash is left ready to run again; finish it so every crash
-        # point converges on the same observable end state.
-        current = _get_plan(restarted.base, plan["id"])
-        if current["state"] != "applied":
-            _apply(restarted.base, current)
-
-        after = _inventory(restarted.base)
-        assert set(after) == set(before), "no asset may be lost or invented by a crash"
-        assert {item["current_sha256"] for item in after.values()} == {
-            item["current_sha256"] for item in before.values()
-        }
-        assert sorted(_content(seeded.lib).values()) == before_bytes
-        assert (seeded.lib / APPROVED).is_dir() and not (seeded.lib / "rome").exists()
-        assert all(APPROVED in item["current_path"] for item in after.values())
-        assert _journal_states(seeded, plan["id"]) == ["complete"]
-    finally:
-        restarted.stop()
-
-
-def test_ambiguous_evidence_is_kept_for_a_human_and_keeps_blocking(tmp_path):
-    seeded = seed_album(tmp_path)
-    first = Server(seeded).start()
-    try:
-        approve_album(first.base, name=APPROVED)
-        plan = _plan(first.base)
-    finally:
-        first.stop()
-
-    _crash_during_apply(seeded, plan, "moving")
-    # Someone creates the destination while the operation is unresolved: now both
-    # paths exist and nothing can tell which one holds the truth.
-    (seeded.lib / APPROVED).mkdir(exist_ok=True)
-
-    restarted = Server(seeded).start()
-    try:
-        assert _recovery(restarted.base)["items"][0]["classification"] == "manual"
-        resolved = httpx.post(
-            f"{restarted.base}/api/v1/rename-recovery/resolve", timeout=TIMEOUT
-        ).json()
-        assert resolved["manual"] == 1 and resolved["resumed"] == 0 and resolved["completed"] == 0
-        # Still blocking, and still nothing guessed.
-        assert _recovery(restarted.base)["blocks_mutation"] is True
-        assert (seeded.lib / "rome").is_dir()
-    finally:
-        restarted.stop()
-
-
-def test_an_unresolved_rename_is_the_cancellation_boundary(tmp_path):
-    """There is no cancel button once a rename starts. The boundary is that nothing
-    else may mutate the library until the interrupted work is resolved."""
-    seeded = seed_album(tmp_path)
-    first = Server(seeded).start()
-    try:
-        approve_album(first.base, name=APPROVED)
-        plan = _plan(first.base)
-    finally:
-        first.stop()
-
-    _crash_during_apply(seeded, plan, "moving")
-
-    restarted = Server(seeded).start()
-    try:
-        assert _recovery(restarted.base)["blocks_mutation"] is True
-
-        refused = httpx.post(f"{restarted.base}/api/v1/albums/proposals", json={}, timeout=TIMEOUT)
-        assert refused.status_code == 409
-        assert refused.json()["error"]["code"] == "rename_recovery_required"
-
-        # Reading stays available throughout — only mutation is paused.
-        assert httpx.get(f"{restarted.base}/api/v1/albums/evidence", timeout=TIMEOUT).status_code
-        assert len(_inventory(restarted.base)) == 2
-    finally:
-        restarted.stop()
-
-
-def test_rollback_returns_an_interrupted_move_to_its_source(tmp_path):
-    seeded = seed_album(tmp_path)
-    first = Server(seeded).start()
-    try:
-        approve_album(first.base, name=APPROVED)
-        before = _inventory(first.base)
-        plan = _plan(first.base)
-    finally:
-        first.stop()
-
-    # Crash after the content moved but before the database caught up: the operation
-    # is still reversible, which is exactly when rollback is defined.
-    _crash_during_apply(seeded, plan, "moved")
-
-    restarted = Server(seeded).start()
-    try:
-        rolled = httpx.post(
-            f"{restarted.base}/api/v1/rename-plans/{plan['id']}/rollback", timeout=TIMEOUT
-        ).json()
-        assert rolled["rolled_back"] == 1 and rolled["state"] == "rolled_back"
-
-        assert (seeded.lib / "rome" / "a.jpg").exists()
-        assert not (seeded.lib / APPROVED).exists()
-        after = _inventory(restarted.base)
-        assert set(after) == set(before)
-        assert all(item["current_path"].endswith(".jpg") for item in after.values())
-        assert _recovery(restarted.base)["blocks_mutation"] is False
-    finally:
-        restarted.stop()
-
-
-# ── durability ───────────────────────────────────────────────────────────────
-
-
-def test_the_applied_state_survives_a_full_restart(tmp_path):
-    seeded = seed_album(tmp_path)
-    first = Server(seeded).start()
-    try:
-        approve_album(first.base, name=APPROVED)
-        plan = _plan(first.base)
-        _apply(first.base, plan, expected_checksum=plan["checksum"]).raise_for_status()
-        expected = _inventory(first.base)
-    finally:
-        first.stop()
-
-    restarted = Server(seeded).start()
-    try:
-        assert _inventory(restarted.base) == expected
-        after = _get_plan(restarted.base, plan["id"])
-        assert after["state"] == "applied"
-        assert after["checksum"] == plan["checksum"], "the plan's evidence is immutable"
-        assert [op["journal_state"] for op in after["operations"]] == ["complete"]
-        assert all(op["verified_at"] for op in after["operations"])
-        assert _recovery(restarted.base) == {"blocks_mutation": False, "items": []}
-    finally:
-        restarted.stop()

tests/e2e/test_renames_ui.py

@@ -11,22 +11,89 @@ Renames really happen here: the assertions read the filesystem afterwards.
 
 from __future__ import annotations
 
+import uuid
+from datetime import datetime, timezone
+
 import httpx
 import pytest
 from playwright.sync_api import expect
 
-from tests.e2e._pipeline_harness import Server, approve_album, seed_album, session_factory
+from tests.e2e._pipeline_harness import Server, image, seed_library
 
 # Part of the Phase D acceptance command (US04-06); mapped to US04-05 for traceability.
 pytestmark = pytest.mark.phase_d
 
+NOW = datetime(2026, 1, 1, tzinfo=timezone.utc)
 TIMEOUT = 10
 APPROVED = "2019 Rome"
 
 
+def _seed(tmp_path):
+    """A library with one album ("rome") whose two photos are SFW and analysed."""
+    seeded = seed_library(tmp_path, {}, {})
+    album = seeded.lib / "rome"
+    album.mkdir()
+    image(album / "a.jpg", 1)
+    image(album / "b.jpg", 2)
+
+    from sqlalchemy import select
+
+    from photo_pipeline.models import AnalysisResult, Asset, SafetyReview
+    from photo_pipeline.services.inventory import InventoryService
+
+    with _factory(seeded) as sf:
+        InventoryService(sf).scan(seeded.lib)
+        with sf() as session:
+            rows = list(session.execute(select(Asset.id, Asset.current_path)).all())
+            for asset_id, path in rows:
+                session.add(
+                    SafetyReview(
+                        id=str(uuid.uuid4()), asset_id=asset_id, decision="sfw", created_at=NOW
+                    )
+                )
+                session.add(
+                    AnalysisResult(
+                        asset_id=asset_id,
+                        status="analyzed",
+                        description=f"a view of {path}",
+                        tags='["ruins", "city"]',
+                        approx_year=2019,
+                        location_hint="Rome",
+                    )
+                )
+            session.commit()
+    return seeded
+
+
+class _factory:
+    """Session factory against the seeded database, for the few things a test has to
+    set up or inspect below the API (journal states, mainly)."""
+
+    def __init__(self, seeded):
+        self._seeded = seeded
+
+    def __enter__(self):
+        from photo_pipeline.config import Config
+        from photo_pipeline.db import create_db_engine, create_session_factory, run_migrations
+
+        config = Config.from_env(
+            {
+                "PHOTO_PIPELINE_DATA_DIR": str(self._seeded.data),
+                "PHOTO_PIPELINE_LIBRARY_ROOTS": str(self._seeded.lib),
+            }
+        )
+        run_migrations(config.database_url)
+        self._engine = create_db_engine(config.database_url)
+        return create_session_factory(self._engine)
+
+    def __exit__(self, *_):
+        self._engine.dispose()
+        return False
+
+
 @pytest.fixture
 def server(tmp_path):
-    seeded = seed_album(tmp_path)
+    seeded = _seed(tmp_path)
     running = Server(seeded).start()
     running.seeded = seeded
     try:
@@ -38,8 +105,22 @@ def server(tmp_path):
 # ── helpers ──────────────────────────────────────────────────────────────────
 
 
-def _approve(base, *, name=APPROVED):
+def _approve(base, *, album="rome", name=APPROVED):
-    approve_album(base, name=name)
+    """Generate a proposal, set its final name, and approve it — the state a rename
+    plan is built from."""
+    httpx.post(f"{base}/api/v1/albums/proposals", json={}, timeout=TIMEOUT).raise_for_status()
+    current = httpx.get(f"{base}/api/v1/albums/proposals/{album}", timeout=TIMEOUT).json()
+    httpx.post(
+        f"{base}/api/v1/albums/proposals/{album}/edit",
+        json={"name": name, "expected_version": current["version"]},
+        timeout=TIMEOUT,
+    ).raise_for_status()
+    current = httpx.get(f"{base}/api/v1/albums/proposals/{album}", timeout=TIMEOUT).json()
+    httpx.post(
+        f"{base}/api/v1/albums/proposals/{album}/approve",
+        json={"expected_version": current["version"]},
+        timeout=TIMEOUT,
+    ).raise_for_status()
 
 
 def _build(base):
@@ -64,7 +145,7 @@ def _interrupt(seeded, plan, *, state="moving", make_destination=False):
     """
     from photo_pipeline.services.rename_journal import RenameJournal
 
-    with session_factory(seeded) as sf:
+    with _factory(seeded) as sf:
         journal = RenameJournal(sf)
         operation = journal.operations(plan["id"])[0]
         journal.begin(operation["id"], worker_id="crashed", fencing_token=1)
@@ -182,22 +263,6 @@ def test_the_applied_plan_survives_a_reload(page, server):
     expect(page.get_by_test_id("apply-result")).to_have_count(0)
 
 
-def test_the_view_matches_the_journal_after_a_server_restart(page, server):
-    _open_plan(page, server)
-    page.get_by_test_id("apply-plan").click()
-    expect(page.get_by_test_id("plan-state")).to_have_text("applied")
-
-    server.stop()
-    server.start()  # same port, so the page reconnects to a genuinely fresh process
-
-    page.reload()
-    expect(page.get_by_test_id("plan-state")).to_have_text("applied")
-    expect(page.get_by_test_id("operation-row").first.get_by_test_id("op-state")).to_have_text(
-        "complete"
-    )
-    expect(page.get_by_test_id("op-verified")).to_be_visible()
-
-
 def test_apply_can_be_confirmed_from_the_keyboard(page, server):
     _open_plan(page, server)
     page.get_by_test_id("apply-plan").focus()
@@ -222,7 +287,7 @@ def test_rollback_returns_an_unfinished_move_to_its_source(page, server):
     # the destination and the operation is still reversible.
     _interrupt(server.seeded, plan, state="moving")
     (server.seeded.lib / "rome").rename(server.seeded.lib / APPROVED)
-    with session_factory(server.seeded) as sf:
+    with _factory(server.seeded) as sf:
         from photo_pipeline.services.rename_journal import RenameJournal
 
         journal = RenameJournal(sf)

tests/e2e/test_traceability.py

@@ -12,19 +12,12 @@ from pathlib import Path
 REPO = Path(__file__).resolve().parents[2]
 MAP = json.loads((REPO / "tests" / "story_traceability.json").read_text())["stories"]
 PHASE_A_STORIES = {f"US01-0{n}" for n in range(1, 8)}
-PHASE_D_STORIES = {f"US04-0{n}" for n in range(1, 7)}
 
 
 def test_all_phase_a_stories_are_mapped():
     assert PHASE_A_STORIES <= set(MAP)
 
 
-def test_all_phase_d_stories_are_mapped():
-    """US04-06 acceptance: every guarded-rename story, plan through browser, is tied
-    to automated tests — renaming is the first thing that mutates the real library."""
-    assert PHASE_D_STORIES <= set(MAP)
-
-
 def test_every_mapped_test_file_exists_and_is_nonempty():
     for story, files in MAP.items():
         assert files, f"{story} maps to no tests"

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

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"

tests/integration/test_upload_reports.py

@@ -1,333 +0,0 @@
-"""Durable per-item upload outcomes (US05-03).
-
-The uploader is a real executable that prints a real report, driven through the
-real batch service, so the classification path exercised here is the one
-production uses. What matters is that the stored outcome is never more optimistic
-than the evidence: an unmentioned file, an unpinned uploader version, or counts
-that disagree all leave the batch requiring verification.
-"""
-
-import json
-import stat
-import threading
-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.integrations import immich_go_report as report_parser
-from photo_pipeline.models import AnalysisResult, Asset, SafetyReview
-from photo_pipeline.services.hashing import sha256_file
-from photo_pipeline.services.upload_batches import BatchState, UploadBatchService
-from photo_pipeline.services.upload_reports import (
-    REQUIRES_VERIFICATION,
-    VERIFIED,
-    UploadReportService,
-)
-from photo_pipeline.services.uploads import UploadService
-
-NOW = datetime(2026, 1, 1, tzinfo=timezone.utc)
-SUPPORTED_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, report_body: str, *, version=SUPPORTED_VERSION):
-    """An uploader that answers ``--version`` and prints ``report_body``."""
-    path = tmp_path / "immich-go"
-    path.write_text(
-        "#!/bin/sh\n"
-        f'if [ "$1" = "--version" ]; then echo "{version}"; exit 0; fi\n'
-        f"cat <<'REPORT'\n{report_body}\nREPORT\n"
-        "exit 0\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, uploader):
-    (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": "immich-sentinel-9f3a2b",
-            "PHOTO_PIPELINE_IMMICH_GO_BINARY": str(uploader),
-        }
-    )
-    run_migrations(config.database_url)
-    return config, create_session_factory(create_db_engine(config.database_url)), lib
-
-
-def _album(sf, lib, album="rome", names=("new.jpg", "dup.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 _run(sf, config, albums=None):
-    """Create a batch from a fresh preflight and run it to completion."""
-    report = UploadService(sf, config=config).preflight(albums)
-    assert report["state"] == "ready", report["blockers"]
-    service = UploadBatchService(sf, config=config)
-    (batch,) = service.create(albums, token=report["token"])
-    return service.run(batch["id"])
-
-
-def _outcomes(batch) -> dict:
-    return {Path(item["path"]).name: item["outcome"] for item in batch["items"]}
-
-
-def _report_for(folder, **outcomes) -> str:
-    """A text-v1 report plus a matching summary line."""
-    lines = {
-        "uploaded": "INFO  uploaded {path}",
-        "duplicate": "INFO  server has the same file {path}",
-        "upgraded": "INFO  server has an older file, upgrading {path}",
-        "failed": "ERROR error uploading {path}: connection reset",
-    }
-    body = [lines[outcome].format(path=folder / name) for name, outcome in outcomes.items()]
-    totals: dict[str, int] = {}
-    for outcome in outcomes.values():
-        totals[outcome] = totals.get(outcome, 0) + 1
-    words = {
-        "uploaded": "Uploaded",
-        "duplicate": "duplicates",
-        "upgraded": "upgraded",
-        "failed": "errors",
-    }
-    body.append(", ".join(f"{words[k]} {v}" for k, v in totals.items()))
-    return "\n".join(body)
-
-
-# ── classification ───────────────────────────────────────────────────────────
-
-
-def test_every_file_gets_its_reported_outcome_with_evidence(tmp_path, immich_server):
-    lib = tmp_path / "lib"
-    report = _report_for(lib / "rome", **{"new.jpg": "uploaded", "dup.jpg": "duplicate"})
-    config, sf, lib = _env(tmp_path, immich_server, _uploader(tmp_path, report))
-    _album(sf, lib)
-
-    batch = _run(sf, config)
-
-    assert batch["state"] == BatchState.SUCCEEDED
-    assert batch["outcome_state"] == VERIFIED
-    assert _outcomes(batch) == {"new.jpg": "uploaded", "dup.jpg": "duplicate"}
-    assert batch["outcome_counts"]["uploaded"] == 1
-    assert batch["outcome_counts"]["duplicate"] == 1
-    assert batch["parser"] == "text-v1"
-    assert batch["parser_version"] == report_parser.PARSER_VERSION
-    assert batch["parsed_at"]
-    for item in batch["items"]:
-        assert item["sha1"] and item["sha256"], "the uploaded bytes stay identifiable"
-        assert Path(item["path"]).name in item["evidence"]
-        assert item["outcome_at"]
-
-
-def test_a_file_the_report_never_mentions_is_unknown_not_sent(tmp_path, immich_server):
-    """Exit code 0 says the process ended well, not that this photo reached Immich."""
-    lib = tmp_path / "lib"
-    report = _report_for(lib / "rome", **{"new.jpg": "uploaded"})
-    config, sf, lib = _env(tmp_path, immich_server, _uploader(tmp_path, report))
-    _album(sf, lib)
-
-    batch = _run(sf, config)
-
-    assert batch["state"] == BatchState.SUCCEEDED, "the process itself succeeded"
-    assert _outcomes(batch)["dup.jpg"] == report_parser.UNKNOWN
-    assert batch["outcome_state"] == REQUIRES_VERIFICATION
-
-
-def test_an_unsupported_uploader_version_makes_every_item_unknown(tmp_path, immich_server):
-    lib = tmp_path / "lib"
-    report = _report_for(lib / "rome", **{"new.jpg": "uploaded", "dup.jpg": "uploaded"})
-    uploader = _uploader(tmp_path, report, version="immich-go 9.99.0")
-    config, sf, lib = _env(tmp_path, immich_server, uploader)
-    _album(sf, lib)
-
-    batch = _run(sf, config)
-
-    assert batch["parser"] is None
-    assert set(_outcomes(batch).values()) == {report_parser.UNKNOWN}
-    assert batch["outcome_state"] == REQUIRES_VERIFICATION
-
-
-def test_an_empty_report_leaves_everything_unknown(tmp_path, immich_server):
-    config, sf, lib = _env(tmp_path, immich_server, _uploader(tmp_path, ""))
-    _album(sf, lib)
-
-    batch = _run(sf, config)
-
-    assert set(_outcomes(batch).values()) == {report_parser.UNKNOWN}
-    assert batch["outcome_state"] == REQUIRES_VERIFICATION
-
-
-def test_a_failed_upload_is_classified_from_its_own_line(tmp_path, immich_server):
-    lib = tmp_path / "lib"
-    report = _report_for(lib / "rome", **{"new.jpg": "uploaded", "dup.jpg": "failed"})
-    config, sf, lib = _env(tmp_path, immich_server, _uploader(tmp_path, report))
-    _album(sf, lib)
-
-    batch = _run(sf, config)
-
-    assert _outcomes(batch) == {"new.jpg": "uploaded", "dup.jpg": "failed"}
-    # Every file is accounted for, so the evidence is complete even though one
-    # upload failed — resolving the failure is US05-04's job, not a re-parse.
-    assert batch["outcome_state"] == VERIFIED
-
-
-# ── reconciliation ───────────────────────────────────────────────────────────
-
-
-def test_a_summary_that_disagrees_with_the_lines_requires_verification(tmp_path, immich_server):
-    lib = tmp_path / "lib"
-    report = (
-        _report_for(lib / "rome", **{"new.jpg": "uploaded", "dup.jpg": "uploaded"}).rsplit("\n", 1)[
-            0
-        ]
-        + "\nUploaded 5"
-    )
-    config, sf, lib = _env(tmp_path, immich_server, _uploader(tmp_path, report))
-    _album(sf, lib)
-
-    batch = _run(sf, config)
-
-    assert batch["report_counts"] == {"uploaded": 5}
-    assert batch["outcome_counts"]["uploaded"] == 2
-    assert batch["outcome_state"] == REQUIRES_VERIFICATION
-
-
-def test_a_reported_file_outside_the_batch_requires_verification(tmp_path, immich_server):
-    """The uploader touched something the batch never approved."""
-    lib = tmp_path / "lib"
-    report = _report_for(
-        lib / "rome", **{"new.jpg": "uploaded", "dup.jpg": "uploaded", "stranger.jpg": "uploaded"}
-    )
-    config, sf, lib = _env(tmp_path, immich_server, _uploader(tmp_path, report))
-    _album(sf, lib)
-
-    batch = _run(sf, config)
-
-    assert set(_outcomes(batch).values()) == {"uploaded"}
-    assert batch["outcome_state"] == REQUIRES_VERIFICATION
-
-
-# ── idempotency and restart ──────────────────────────────────────────────────
-
-
-def test_reprocessing_the_same_report_is_idempotent(tmp_path, immich_server):
-    lib = tmp_path / "lib"
-    report = _report_for(lib / "rome", **{"new.jpg": "uploaded", "dup.jpg": "duplicate"})
-    config, sf, lib = _env(tmp_path, immich_server, _uploader(tmp_path, report))
-    _album(sf, lib)
-    batch = _run(sf, config)
-
-    first = UploadReportService(sf).ingest(batch["id"])
-    second = UploadReportService(sf).ingest(batch["id"])
-
-    assert first == second
-    reread = UploadBatchService(sf, config=config).get(batch["id"])
-    assert _outcomes(reread) == _outcomes(batch)
-    assert len(reread["items"]) == 2, "re-import must not duplicate item rows"
-    assert reread["outcome_counts"] == batch["outcome_counts"]
-
-
-def test_outcomes_survive_a_restart_and_reach_the_api(tmp_path, immich_server):
-    lib = tmp_path / "lib"
-    report = _report_for(lib / "rome", **{"new.jpg": "uploaded", "dup.jpg": "duplicate"})
-    config, sf, lib = _env(tmp_path, immich_server, _uploader(tmp_path, report))
-    _album(sf, lib)
-    batch = _run(sf, config)
-
-    with TestClient(create_app(config)) as client:  # a fresh application process
-        fetched = client.get(f"/api/v1/upload-batches/{batch['id']}").json()
-
-    assert fetched["outcome_state"] == VERIFIED
-    assert {Path(i["path"]).name: i["outcome"] for i in fetched["items"]} == {
-        "new.jpg": "uploaded",
-        "dup.jpg": "duplicate",
-    }
-    assert json.dumps(fetched)  # the record stays JSON-serialisable for the UI
-
-
-def test_a_missing_report_file_does_not_lose_the_batch(tmp_path, immich_server):
-    lib = tmp_path / "lib"
-    report = _report_for(lib / "rome", **{"new.jpg": "uploaded", "dup.jpg": "duplicate"})
-    config, sf, lib = _env(tmp_path, immich_server, _uploader(tmp_path, report))
-    _album(sf, lib)
-    batch = _run(sf, config)
-    Path(batch["report_path"]).unlink()
-
-    result = UploadReportService(sf).ingest(batch["id"])
-
-    assert result["outcome_state"] == REQUIRES_VERIFICATION
-    assert set(_outcomes(UploadBatchService(sf, config=config).get(batch["id"])).values()) == {
-        report_parser.UNKNOWN
-    }
-
-
-def test_ingesting_an_unknown_batch_is_an_error(tmp_path, immich_server):
-    config, sf, lib = _env(tmp_path, immich_server, _uploader(tmp_path, ""))
-
-    with pytest.raises(KeyError):
-        UploadReportService(sf).ingest("does-not-exist")

tests/integration/test_upload_verification.py

@@ -1,535 +0,0 @@
-"""Verifying, retrying, and resolving uncertain uploads (US05-04).
-
-The Immich boundary is a real HTTP server here: it answers ``/api/server/ping``
-and ``/api/assets/bulk-upload-check`` exactly as the app's own adapter parses
-them, and the set of checksums it "holds" is what each fault scenario controls.
-The uploader stays a real executable driven through the real batch service, so
-every state under test is reached the way production reaches it.
-
-The invariant these tests defend is one-directional: uncertainty may only become
-success when something authoritative said so — the server, or an operator who
-recorded what they checked.
-"""
-
-import json
-import stat
-import threading
-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.models import AnalysisResult, Asset, SafetyReview, UploadBatch
-from photo_pipeline.services.hashing import sha256_file
-from photo_pipeline.services.upload_batches import BatchError, BatchState, UploadBatchService
-from photo_pipeline.services.upload_reports import REQUIRES_VERIFICATION, VERIFIED
-from photo_pipeline.services.upload_verification import (
-    ABSENT,
-    INCONCLUSIVE,
-    MANUAL,
-    PRESENT,
-    UploadVerificationService,
-    VerificationError,
-    retry_blockers,
-)
-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 Immich ──────────────────────────────────────────────────────────────
-
-
-class _FakeImmich:
-    """The server's view of the world: which checksums it holds, and whether it
-    is willing to answer at all."""
-
-    def __init__(self) -> None:
-        self.held: set[str] = set()
-        self.available = True
-        self.checked: list[str] = []
-
-
-def _handler(state: _FakeImmich):
-    class Handler(BaseHTTPRequestHandler):
-        def do_GET(self):  # noqa: N802 (BaseHTTPRequestHandler API)
-            self._json(200, {"res": "pong"})
-
-        def do_POST(self):  # noqa: N802
-            body = json.loads(self.rfile.read(int(self.headers["Content-Length"] or 0)) or "{}")
-            if not state.available:
-                self._json(503, {"error": "unavailable"})
-                return
-            if self.headers.get("x-api-key") != SENTINEL_KEY:
-                self._json(401, {"error": "unauthorized"})
-                return
-            results = []
-            for asset in body.get("assets", []):
-                state.checked.append(asset["checksum"])
-                results.append(
-                    {"id": asset["id"], "action": "reject", "reason": "duplicate"}
-                    if asset["checksum"] in state.held
-                    else {"id": asset["id"], "action": "accept"}
-                )
-            self._json(200, {"results": results})
-
-        def _json(self, status: int, payload: dict) -> None:
-            body = json.dumps(payload).encode()
-            self.send_response(status)
-            self.send_header("Content-Type", "application/json")
-            self.send_header("Content-Length", str(len(body)))
-            self.end_headers()
-            self.wfile.write(body)
-
-        def log_message(self, *args):
-            pass
-
-    return Handler
-
-
-@pytest.fixture
-def immich():
-    state = _FakeImmich()
-    server = HTTPServer(("127.0.0.1", 0), _handler(state))
-    threading.Thread(target=server.serve_forever, daemon=True).start()
-    state.url = f"http://127.0.0.1:{server.server_port}"
-    yield state
-    server.shutdown()
-    server.server_close()
-
-
-# ── environment ──────────────────────────────────────────────────────────────
-
-
-def _uploader(tmp_path, report_body: str = "", *, exit_code: int = 0, version=UPLOADER_VERSION):
-    path = tmp_path / "immich-go"
-    path.write_text(
-        "#!/bin/sh\n"
-        f'if [ "$1" = "--version" ]; then echo "{version}"; exit 0; fi\n'
-        f"cat <<'REPORT'\n{report_body}\nREPORT\n"
-        f"exit {exit_code}\n"
-    )
-    path.chmod(path.stat().st_mode | stat.S_IEXEC | stat.S_IXGRP | stat.S_IXOTH)
-    return path
-
-
-def _env(tmp_path, immich, 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": immich.url,
-            "PHOTO_PIPELINE_IMMICH_API_KEY": SENTINEL_KEY,
-            "PHOTO_PIPELINE_IMMICH_GO_BINARY": str(
-                uploader if uploader is not None else _uploader(tmp_path)
-            ),
-        }
-    )
-    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 _batch(sf, config, albums=None):
-    report = UploadService(sf, config=config).preflight(albums)
-    assert report["state"] == "ready", report["blockers"]
-    (batch,) = UploadBatchService(sf, config=config).create(albums, token=report["token"])
-    return batch
-
-
-def _interrupt(sf, config, batch_id):
-    """Leave behind exactly what a killed worker leaves: a mid-flight attempt."""
-    with sf() as session:
-        session.get(UploadBatch, batch_id).state = BatchState.RUNNING
-        session.commit()
-    UploadBatchService(sf, config=config).recover()
-
-
-def _server_holds(immich, batch):
-    immich.held.update(item["sha1"] for item in batch["items"])
-
-
-def _by_name(batch) -> dict:
-    return {Path(item["path"]).name: item for item in batch["items"]}
-
-
-def _uncertain_batch(tmp_path, immich, *, hold: bool):
-    """A batch whose attempt died mid-upload, with the server holding the bytes or not."""
-    config, sf, lib = _env(tmp_path, immich)
-    _album(sf, lib)
-    batch = _batch(sf, config)
-    if hold:
-        _server_holds(immich, batch)
-    _interrupt(sf, config, batch["id"])
-    return config, sf, lib, batch
-
-
-# ── accepted-but-unknown ─────────────────────────────────────────────────────
-
-
-def test_acceptance_then_lost_response_is_confirmed_by_the_server(tmp_path, immich):
-    """The classic lost response: Immich took the files, the app never saw a report."""
-    config, sf, lib, batch = _uncertain_batch(tmp_path, immich, hold=True)
-
-    result = UploadVerificationService(sf, config=config).verify(batch["id"])
-
-    assert result["state"] == BatchState.SUCCEEDED, "server evidence resolves the uncertainty"
-    assert result["outcome_state"] == VERIFIED
-    assert result["counts"] == {PRESENT: 2}
-    verified = UploadBatchService(sf, config=config).get(batch["id"])
-    for item in verified["items"]:
-        assert item["outcome"] == "uploaded"
-        assert item["verification"] == PRESENT
-        assert item["sha1"] in item["evidence"], "the exact bytes are named in the evidence"
-    assert immich.checked, "verification actually asked the server"
-
-
-def test_timeout_before_acceptance_leaves_a_safe_retry(tmp_path, immich):
-    """Nothing arrived, so the batch becomes a plain failure and may run again."""
-    config, sf, lib, batch = _uncertain_batch(tmp_path, immich, hold=False)
-    service = UploadBatchService(sf, config=config)
-
-    result = UploadVerificationService(sf, config=config).verify(batch["id"])
-
-    assert result["state"] == BatchState.FAILED
-    assert result["counts"] == {ABSENT: 2}
-    assert {item["outcome"] for item in service.get(batch["id"])["items"]} == {"failed"}
-    assert retry_blockers(service.get(batch["id"])) == []
-    assert service.run(batch["id"])["state"] == BatchState.SUCCEEDED
-
-
-def test_an_uncertain_batch_cannot_be_retried_before_verification(tmp_path, immich):
-    config, sf, lib, batch = _uncertain_batch(tmp_path, immich, hold=True)
-    service = UploadBatchService(sf, config=config)
-
-    with pytest.raises(BatchError) as error:
-        service.run(batch["id"])
-
-    assert error.value.code == "requires_verification"
-    with TestClient(create_app(config)) as client:
-        response = client.post(f"/api/v1/upload-batches/{batch['id']}/start")
-    assert response.status_code == 409
-    assert response.json()["error"]["code"] == "requires_verification"
-
-
-def test_a_partly_arrived_batch_is_a_failure_not_a_success(tmp_path, immich):
-    config, sf, lib = _env(tmp_path, immich)
-    _album(sf, lib)
-    batch = _batch(sf, config)
-    immich.held.add(batch["items"][0]["sha1"])  # only one file made it
-    _interrupt(sf, config, batch["id"])
-
-    result = UploadVerificationService(sf, config=config).verify(batch["id"])
-
-    assert result["state"] == BatchState.FAILED
-    assert result["counts"] == {PRESENT: 1, ABSENT: 1}
-    assert result["outcome_state"] == VERIFIED, "every file is accounted for"
-
-
-# ── parser uncertainty ───────────────────────────────────────────────────────
-
-
-def test_parser_uncertainty_is_settled_by_server_evidence(tmp_path, immich):
-    """An unpinned uploader version leaves every item unknown; the server decides."""
-    config, sf, lib = _env(tmp_path, immich, _uploader(tmp_path, "done", version="immich-go 9.9.9"))
-    _album(sf, lib)
-    batch = _batch(sf, config)
-    service = UploadBatchService(sf, config=config)
-    ran = service.run(batch["id"])
-    assert ran["state"] == BatchState.SUCCEEDED and ran["outcome_state"] == REQUIRES_VERIFICATION
-    _server_holds(immich, ran)
-
-    result = UploadVerificationService(sf, config=config).verify(batch["id"])
-
-    assert result["outcome_state"] == VERIFIED
-    assert {item["outcome"] for item in service.get(batch["id"])["items"]} == {"uploaded"}
-
-
-def test_an_unreachable_server_never_turns_uncertainty_into_success(tmp_path, immich):
-    config, sf, lib, batch = _uncertain_batch(tmp_path, immich, hold=True)
-    immich.available = False
-    service = UploadBatchService(sf, config=config)
-
-    result = UploadVerificationService(sf, config=config).verify(batch["id"])
-
-    assert result["server_reachable"] is False
-    assert result["detail"], "the reason the server could not answer is reported"
-    assert result["counts"] == {INCONCLUSIVE: 2}
-    assert result["state"] == BatchState.UNKNOWN, "still uncertain, not succeeded"
-    assert result["outcome_state"] == REQUIRES_VERIFICATION
-    assert [b["code"] for b in retry_blockers(service.get(batch["id"]))] == [
-        "requires_verification"
-    ]
-
-
-# ── safe failures ────────────────────────────────────────────────────────────
-
-
-def test_a_plain_uploader_failure_is_retryable_without_verification(tmp_path, immich):
-    """Nothing uncertain happened: the process failed before/while reporting an error."""
-    config, sf, lib = _env(tmp_path, immich, _uploader(tmp_path, "boom", exit_code=1))
-    _album(sf, lib)
-    batch = _batch(sf, config)
-    service = UploadBatchService(sf, config=config)
-    assert service.run(batch["id"])["state"] == BatchState.FAILED
-
-    assert retry_blockers(service.get(batch["id"])) == []
-    with TestClient(create_app(config)) as client:
-        assert client.post(f"/api/v1/upload-batches/{batch['id']}/start").status_code == 200
-
-
-# ── changed bytes ────────────────────────────────────────────────────────────
-
-
-def test_bytes_changed_after_upload_warn_and_block_a_rerun(tmp_path, immich):
-    config, sf, lib = _env(tmp_path, immich)
-    folder = _album(sf, lib)
-    batch = _batch(sf, config)
-    service = UploadBatchService(sf, config=config)
-    ran = service.run(batch["id"])
-    _server_holds(immich, ran)
-    (folder / "a.jpg").write_bytes(b"edited after the upload")
-
-    result = UploadVerificationService(sf, config=config).verify(batch["id"])
-
-    assert result["stale_bytes"] is True
-    changed = _by_name(result)["a.jpg"]
-    assert changed["changed_after_upload"] is True
-    assert _by_name(result)["b.jpg"]["changed_after_upload"] is False
-    stored = service.get(batch["id"])
-    assert stored["stale_bytes"] is True, "the warning is durable, not only in the response"
-    assert _by_name(stored)["a.jpg"]["observed_sha256"] != _by_name(stored)["a.jpg"]["sha256"]
-    with pytest.raises(BatchError) as error:
-        service.run(batch["id"])
-    assert error.value.code == "changed_after_upload"
-    with TestClient(create_app(config)) as client:
-        response = client.post(f"/api/v1/upload-batches/{batch['id']}/start")
-    assert response.status_code == 409
-    assert response.json()["error"]["code"] == "changed_after_upload"
-
-
-def test_a_deleted_file_counts_as_changed_after_upload(tmp_path, immich):
-    config, sf, lib = _env(tmp_path, immich)
-    folder = _album(sf, lib)
-    batch = _batch(sf, config)
-    ran = UploadBatchService(sf, config=config).run(batch["id"])
-    _server_holds(immich, ran)
-    (folder / "a.jpg").unlink()
-
-    result = UploadVerificationService(sf, config=config).verify(batch["id"])
-
-    assert _by_name(result)["a.jpg"]["changed_after_upload"] is True
-    assert result["stale_bytes"] is True
-    # The bytes are still on the server: verification is about the upload, not the
-    # local file's continued existence.
-    assert _by_name(result)["a.jpg"]["verification"] == PRESENT
-
-
-# ── repeated verification and history ────────────────────────────────────────
-
-
-def test_repeated_verification_converges_and_keeps_every_answer(tmp_path, immich):
-    config, sf, lib, batch = _uncertain_batch(tmp_path, immich, hold=True)
-    service = UploadVerificationService(sf, config=config)
-
-    first = service.verify(batch["id"])
-    second = service.verify(batch["id"])
-
-    assert first["counts"] == second["counts"] == {PRESENT: 2}
-    assert first["state"] == second["state"] == BatchState.SUCCEEDED
-    history = service.history(batch["id"])
-    assert len(history) == 4, "the audit trail appends, it never overwrites"
-    assert {entry["source"] for entry in history} == {"immich_api"}
-    assert all(entry["action"] == "verify" for entry in history)
-
-
-def test_verification_that_changes_its_mind_keeps_both_answers(tmp_path, immich):
-    """A file that was absent and later present must show both, in order."""
-    config, sf, lib, batch = _uncertain_batch(tmp_path, immich, hold=False)
-    service = UploadVerificationService(sf, config=config)
-    service.verify(batch["id"])
-    _server_holds(immich, batch)
-
-    service.verify(batch["id"])
-
-    asset_id = batch["items"][0]["asset_id"]
-    results = [e["result"] for e in service.history(batch["id"]) if e["asset_id"] == asset_id]
-    assert results == [ABSENT, PRESENT]
-
-
-# ── manual resolution ────────────────────────────────────────────────────────
-
-
-def test_manual_resolution_requires_evidence_and_an_author(tmp_path, immich):
-    config, sf, lib, batch = _uncertain_batch(tmp_path, immich, hold=True)
-    service = UploadVerificationService(sf, config=config)
-    asset_id = batch["items"][0]["asset_id"]
-
-    for kwargs, code in (
-        ({"evidence": "   ", "actor": "dom"}, "evidence_required"),
-        ({"evidence": "checked in Immich", "actor": ""}, "actor_required"),
-        ({"evidence": "checked", "actor": "dom", "outcome": "definitely-fine"}, "invalid_outcome"),
-    ):
-        with pytest.raises(VerificationError) as error:
-            service.resolve(batch["id"], asset_id, **{"outcome": "uploaded", **kwargs})
-        assert error.value.code == code
-
-    assert service.history(batch["id"]) == [], "a refused resolution records nothing"
-    assert UploadBatchService(sf, config=config).get(batch["id"])["state"] == BatchState.UNKNOWN
-
-
-def test_manual_resolution_is_recorded_as_operator_evidence(tmp_path, immich):
-    """An operator may settle what the server cannot — but never anonymously."""
-    config, sf, lib, batch = _uncertain_batch(tmp_path, immich, hold=True)
-    immich.available = False
-    service = UploadVerificationService(sf, config=config)
-    service.verify(batch["id"])  # inconclusive: the server is down
-
-    for item in batch["items"]:
-        result = service.resolve(
-            batch["id"],
-            item["asset_id"],
-            outcome="uploaded",
-            evidence="found in Immich by checksum in the web UI",
-            actor="dom",
-        )
-
-    assert result["state"] == BatchState.SUCCEEDED
-    assert result["outcome_state"] == VERIFIED
-    resolutions = [e for e in service.history(batch["id"]) if e["action"] == "resolve"]
-    assert len(resolutions) == 2
-    assert {e["actor"] for e in resolutions} == {"dom"}
-    assert {e["source"] for e in resolutions} == {"operator"}
-    assert all("web UI" in e["evidence"] for e in resolutions)
-    stored = UploadBatchService(sf, config=config).get(batch["id"])
-    assert {item["verification"] for item in stored["items"]} == {MANUAL}, (
-        "a manual answer stays distinguishable from server evidence"
-    )
-
-
-def test_resolving_one_item_leaves_the_batch_uncertain(tmp_path, immich):
-    config, sf, lib, batch = _uncertain_batch(tmp_path, immich, hold=True)
-    immich.available = False
-    service = UploadVerificationService(sf, config=config)
-    service.verify(batch["id"])
-
-    result = service.resolve(
-        batch["id"],
-        batch["items"][0]["asset_id"],
-        outcome="uploaded",
-        evidence="visible in Immich",
-        actor="dom",
-    )
-
-    assert result["state"] == BatchState.UNKNOWN
-    assert result["outcome_state"] == REQUIRES_VERIFICATION
-
-
-def test_resolving_an_unknown_item_or_batch_is_refused(tmp_path, immich):
-    config, sf, lib, batch = _uncertain_batch(tmp_path, immich, hold=True)
-    service = UploadVerificationService(sf, config=config)
-
-    for batch_id, asset_id in ((batch["id"], "not-in-this-batch"), ("no-such-batch", "x")):
-        with pytest.raises(VerificationError) as error:
-            service.resolve(batch_id, asset_id, outcome="uploaded", evidence="checked", actor="dom")
-        assert error.value.code == "not_found"
-
-
-def test_verifying_an_unknown_batch_is_refused(tmp_path, immich):
-    config, sf, lib = _env(tmp_path, immich)
-
-    with pytest.raises(VerificationError) as error:
-        UploadVerificationService(sf, config=config).verify("does-not-exist")
-
-    assert error.value.code == "not_found"
-
-
-# ── API surface and durability ───────────────────────────────────────────────
-
-
-def test_the_api_verifies_resolves_and_lists_history_without_secrets(tmp_path, immich):
-    config, sf, lib, batch = _uncertain_batch(tmp_path, immich, hold=False)
-
-    with TestClient(create_app(config)) as client:
-        verified = client.post(f"/api/v1/upload-batches/{batch['id']}/verify")
-        resolved = client.post(
-            f"/api/v1/upload-batches/{batch['id']}/resolve",
-            json={
-                "asset_id": batch["items"][0]["asset_id"],
-                "outcome": "skipped",
-                "evidence": "the file was withdrawn from the album",
-                "actor": "dom",
-            },
-        )
-        refused = client.post(
-            f"/api/v1/upload-batches/{batch['id']}/resolve",
-            json={
-                "asset_id": batch["items"][0]["asset_id"],
-                "outcome": "uploaded",
-                "evidence": "",
-                "actor": "dom",
-            },
-        )
-        history = client.get(f"/api/v1/upload-batches/{batch['id']}/verifications")
-        missing = client.post("/api/v1/upload-batches/does-not-exist/verify")
-
-    assert verified.status_code == 200 and verified.json()["counts"] == {ABSENT: 2}
-    assert resolved.status_code == 200
-    assert refused.status_code == 422 and refused.json()["error"]["code"] == "evidence_required"
-    assert len(history.json()["verifications"]) == 3
-    assert missing.status_code == 404
-    assert SENTINEL_KEY not in verified.text + resolved.text + history.text
-
-
-def test_verification_survives_a_restart(tmp_path, immich):
-    config, sf, lib, batch = _uncertain_batch(tmp_path, immich, hold=True)
-    UploadVerificationService(sf, config=config).verify(batch["id"])
-
-    with TestClient(create_app(config)) as client:  # a fresh application process
-        fetched = client.get(f"/api/v1/upload-batches/{batch['id']}").json()
-        history = client.get(f"/api/v1/upload-batches/{batch['id']}/verifications").json()
-
-    assert fetched["state"] == BatchState.SUCCEEDED
-    assert fetched["verified_at"]
-    assert {item["verification"] for item in fetched["items"]} == {PRESENT}
-    assert len(history["verifications"]) == 2
-    assert json.dumps(fetched)  # the record stays JSON-serialisable for the UI

tests/story_traceability.json

@@ -100,23 +100,6 @@
     ],
     "US04-05": [
       "tests/e2e/test_renames_ui.py"
-    ],
-    "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"
-    ],
-    "US05-03": [
-      "tests/unit/test_immich_go_report.py",
-      "tests/integration/test_upload_reports.py"
-    ],
-    "US05-04": [
-      "tests/unit/test_immich_bulk_check.py",
-      "tests/integration/test_upload_verification.py"
     ]
   }
 }

tests/unit/test_immich_bulk_check.py

@@ -1,27 +0,0 @@
-"""How one bulk-upload-check result is read (US05-04).
-
-Only a duplicate rejection proves Immich holds the bytes. Every other answer — a
-rejection for another reason, an action this adapter does not know — must stay
-uncertain, because "the server did not say yes" is not "the file is not there".
-"""
-
-from photo_pipeline.integrations.immich_go import _holds_bytes, bulk_upload_check
-
-
-def test_a_duplicate_rejection_is_the_only_proof_of_possession():
-    assert _holds_bytes({"action": "reject", "reason": "duplicate"}) is True
-    assert _holds_bytes({"action": "accept"}) is False
-
-
-def test_any_other_answer_is_uncertain_never_absent():
-    assert _holds_bytes({"action": "reject", "reason": "unsupported-format"}) is None
-    assert _holds_bytes({"action": "quarantine"}) is None
-    assert _holds_bytes({}) is None
-
-
-def test_missing_credentials_are_reported_not_silently_treated_as_absence():
-    result = bulk_upload_check("", None, {"asset": "abc"})
-
-    assert result["reachable"] is False
-    assert result["present"] == {}
-    assert "credentials" in result["detail"]

tests/unit/test_immich_go_report.py

@@ -1,139 +0,0 @@
-"""Golden parser fixtures for immich-go reports (US05-03).
-
-The reports below are the pinned grammar for each supported uploader version. If a
-future immich-go changes its wording, the fix is a new adapter and a new golden
-report — never a looser pattern here, because a loose pattern is how an unread line
-becomes a false success.
-"""
-
-import pytest
-
-from photo_pipeline.integrations import immich_go_report as parser
-
-TEXT_REPORT = """
-Scanning /lib/rome
-INFO  uploaded /lib/rome/new.jpg
-INFO  server has the same file /lib/rome/dup.jpg
-INFO  server has an older file, upgrading /lib/rome/old.jpg
-WARN  discarded /lib/rome/notes.txt: unsupported file type
-ERROR error uploading /lib/rome/broken.jpg: connection reset by peer
-Uploaded 1, upgraded 1, duplicates 1, skipped 1, errors 1
-"""
-
-JSON_REPORT = """
-{"time":"2026-01-01T10:00:00Z","level":"INFO","msg":"uploaded","file":"/lib/rome/new.jpg"}
-{"level":"INFO","msg":"server has the same file","file":"/lib/rome/dup.jpg"}
-{"level":"INFO","msg":"server has an older file, upgrading","file":"/lib/rome/old.jpg"}
-{"level":"WARN","msg":"discarded: unsupported file type","file":"/lib/rome/notes.txt"}
-{"level":"ERROR","msg":"error uploading","file":"/lib/rome/broken.jpg"}
-{"level":"INFO","msg":"report","counts":{"uploaded":1,"upgraded":1,"duplicates":1,\
-"skipped":1,"errors":1}}
-"""
-
-EXPECTED = {
-    "/lib/rome/new.jpg": parser.UPLOADED,
-    "/lib/rome/dup.jpg": parser.DUPLICATE,
-    "/lib/rome/old.jpg": parser.UPGRADED,
-    "/lib/rome/notes.txt": parser.SKIPPED,
-    "/lib/rome/broken.jpg": parser.FAILED,
-}
-EXPECTED_COUNTS = {
-    parser.UPLOADED: 1,
-    parser.UPGRADED: 1,
-    parser.DUPLICATE: 1,
-    parser.SKIPPED: 1,
-    parser.FAILED: 1,
-}
-
-
-def _outcomes(parsed) -> dict:
-    return {entry["path"]: entry["outcome"] for entry in parsed["entries"]}
-
-
-@pytest.mark.parametrize(
-    ("version", "report", "expected_parser"),
-    [
-        ("immich-go 0.21.0", TEXT_REPORT, "text-v1"),
-        ("immich-go 0.22.3", TEXT_REPORT, "text-v1"),
-        ("immich-go 0.23.1", JSON_REPORT, "json-v1"),
-        ("immich-go 0.24.0", JSON_REPORT, "json-v1"),
-    ],
-)
-def test_every_supported_version_classifies_every_outcome(version, report, expected_parser):
-    parsed = parser.parse(report, version)
-
-    assert parsed["parser"] == expected_parser and parsed["supported"]
-    assert parsed["parser_version"] == parser.PARSER_VERSION
-    assert _outcomes(parsed) == EXPECTED
-    assert parsed["counts"] == EXPECTED_COUNTS
-
-
-def test_an_unsupported_version_parses_nothing_at_all():
-    """A build whose output was never pinned must not be read optimistically."""
-    parsed = parser.parse(TEXT_REPORT, "immich-go 9.99.0")
-
-    assert parsed["parser"] is None and parsed["supported"] is False
-    assert parsed["entries"] == [] and parsed["counts"] is None
-
-
-@pytest.mark.parametrize("version", [None, "", "immich-go dev", "unknown"])
-def test_a_missing_or_unreadable_version_is_unsupported(version):
-    assert parser.parser_for(version) is None
-
-
-def test_malformed_lines_are_counted_never_classified():
-    report = (
-        "{not json at all\n"
-        '{"level":"INFO","msg":"uploaded"}\n'  # no file
-        '["not","an","object"]\n'
-        '{"level":"INFO","msg":"uploaded","file":"/lib/rome/new.jpg"}\n'
-    )
-
-    parsed = parser.parse(report, "immich-go 0.23.1")
-
-    assert _outcomes(parsed) == {"/lib/rome/new.jpg": parser.UPLOADED}
-    assert parsed["unparsed"] == 3
-
-
-def test_a_line_with_a_path_but_no_known_verb_is_not_an_outcome():
-    parsed = parser.parse("INFO  considering /lib/rome/a.jpg\n", "immich-go 0.21.0")
-
-    assert parsed["entries"] == [] and parsed["unparsed"] == 1
-
-
-def test_the_album_name_never_classifies_the_line():
-    """A folder called "errors" must not turn a successful upload into a failure."""
-    parsed = parser.parse("INFO  uploaded /lib/errors/a.jpg\n", "immich-go 0.21.0")
-
-    assert _outcomes(parsed) == {"/lib/errors/a.jpg": parser.UPLOADED}
-
-
-def test_paths_with_spaces_and_reasons_are_read_whole():
-    report = "ERROR error uploading /lib/summer holiday/a b.jpg: connection reset\n"
-
-    parsed = parser.parse(report, "immich-go 0.21.0")
-
-    assert _outcomes(parsed) == {"/lib/summer holiday/a b.jpg": parser.FAILED}
-
-
-def test_evidence_is_bounded_per_entry():
-    line = "INFO  uploaded /lib/rome/a.jpg " + "x" * 5_000
-
-    (entry,) = parser.parse(line, "immich-go 0.21.0")["entries"]
-
-    assert len(entry["evidence"]) == parser.MAX_EVIDENCE_CHARS
-
-
-def test_a_looping_uploader_cannot_produce_unbounded_entries(monkeypatch):
-    monkeypatch.setattr(parser, "MAX_ENTRIES", 10)
-    report = "".join(f"INFO  uploaded /lib/rome/{n}.jpg\n" for n in range(50))
-
-    parsed = parser.parse(report, "immich-go 0.21.0")
-
-    assert len(parsed["entries"]) == 10 and parsed["entries_truncated"] is True
-
-
-def test_a_report_without_a_summary_reports_no_counts():
-    parsed = parser.parse("INFO  uploaded /lib/rome/a.jpg\n", "immich-go 0.21.0")
-
-    assert parsed["counts"] is None, "counts are read, never invented"