US05-04: Verify, Retry, and Resolve Uncertain Uploads

photo_pipeline/api/routes/uploads.py

@@ -1,9 +1,12 @@
-"""Upload preflight and batch API (US05-01, US05-02).
+"""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
@@ -19,6 +22,11 @@ from photo_pipeline.services.upload_batches import (
     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"])
@@ -36,6 +44,16 @@ class CreateBatchRequest(PreflightRequest):
     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)
 
@@ -44,6 +62,12 @@ 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}})
 
@@ -61,9 +85,11 @@ def preflight(request: Request, body: PreflightRequest | None = None):
 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
-        )}
+        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:
@@ -92,6 +118,11 @@ def start_batch(batch_id: str, request: 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,
@@ -105,6 +136,38 @@ def start_batch(batch_id: str, request: Request):
     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:

photo_pipeline/integrations/immich_go.py

@@ -8,7 +8,9 @@ activity log. Both come from the same builder so the preview can never drift fro
 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.
+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
@@ -35,6 +37,11 @@ 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
@@ -89,6 +96,68 @@ def ping(server_url: str, *, timeout: float = PING_TIMEOUT_SECONDS) -> tuple[boo
     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,

photo_pipeline/models/__init__.py

@@ -14,7 +14,7 @@ from photo_pipeline.models.duplicates import (
 from photo_pipeline.models.jobs import Job, JobEvent, JobItem
 from photo_pipeline.models.renames import RenameOperation, RenamePlan
 from photo_pipeline.models.thumbnails import Thumbnail
-from photo_pipeline.models.uploads import UploadBatch, UploadItem
+from photo_pipeline.models.uploads import UploadBatch, UploadItem, UploadVerification
 from photo_pipeline.models.workflow import AnalysisResult, SafetyReview
 
 __all__ = [
@@ -32,6 +32,7 @@ __all__ = [
     "Thumbnail",
     "UploadBatch",
     "UploadItem",
+    "UploadVerification",
     "SafetyReview",
     "AnalysisResult",
 ]

photo_pipeline/models/uploads.py

@@ -69,6 +69,11 @@ class UploadBatch(Base):
     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)
 
@@ -102,6 +107,45 @@ class UploadItem(Base):
     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/upload_batches.py

@@ -16,7 +16,8 @@ cannot undo, so the discipline is:
 - **an interrupted attempt is uncertain, not failed.** Immich may have accepted
   files the app never saw a report for, so ``recover`` marks a batch whose worker
   vanished ``unknown_requires_verification`` (concept §15) instead of retrying it
-  blindly. Resolving that is US05-04.
+  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.
@@ -40,6 +41,7 @@ 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
 
 
@@ -62,6 +64,7 @@ RUNNABLE_STATES = frozenset({BatchState.PLANNED, BatchState.FAILED, BatchState.C
 # 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
@@ -162,13 +165,11 @@ class UploadBatchService:
     def run(self, batch_id: str, *, worker_id: str = "uploader", cancelled=None) -> dict:
         """Run one batch to completion. Blocks for the duration of the upload."""
         batch = self._require(batch_id)
-        if batch["state"] not in RUNNABLE_STATES:
-            code = (
-                "requires_verification"
-                if batch["state"] == BatchState.UNKNOWN
-                else "not_runnable"
-            )
-            raise BatchError(code, f"batch {batch_id} is {batch['state']}")
+        # 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")
@@ -189,9 +190,7 @@ class UploadBatchService:
 
         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"
-        )
+        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,
@@ -426,6 +425,10 @@ def _batch_dict(row: UploadBatch, items: list[UploadItem]) -> dict:
         "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": [
@@ -438,6 +441,10 @@ def _batch_dict(row: UploadBatch, items: list[UploadItem]) -> dict:
                 "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_verification.py

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