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

migrations/versions/0009_upload_report_outcomes.py

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

@@ -0,0 +1,70 @@
+"""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/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/integrations/immich_go_report.py

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

@@ -57,6 +57,23 @@ class UploadBatch(Base):
     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)
 
@@ -81,10 +98,54 @@ class UploadItem(Base):
     # Hashes of the bytes as they were when the batch was created.
     sha256: Mapped[str | None] = mapped_column(String)
     sha1: Mapped[str | None] = mapped_column(String)
-    # pending | sent | failed — the per-asset upload *result* is parsed from the
-    # report in US05-03 and verified in US05-04; ``sent`` only means the batch
-    # process exited successfully.
+    # 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/upload_batches.py

@@ -16,12 +16,14 @@ 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 deliberately not interpreted here — parsing the
-report is US05-03. What this story guarantees is that the report exists, is bounded,
-carries no credential, and that the batch's state is an honest description of what
-the process did.
+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
@@ -38,6 +40,8 @@ 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
 
 
@@ -60,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
@@ -160,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")
@@ -187,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,
@@ -238,6 +239,13 @@ class UploadBatchService:
         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:
@@ -409,6 +417,18 @@ def _batch_dict(row: UploadBatch, items: list[UploadItem]) -> dict:
         "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": [
@@ -418,6 +438,13 @@ def _batch_dict(row: UploadBatch, items: list[UploadItem]) -> dict:
                 "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

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

@@ -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
+        ],
+    }

tests/integration/test_upload_reports.py

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

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

@@ -109,6 +109,14 @@
     ],
     "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

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

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