US05-03: Parse and Persist Uploader Outcomes

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)

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/uploads.py

@@ -57,6 +57,18 @@ 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
+
     error_code: Mapped[str | None] = mapped_column(String)
     error_message: Mapped[str | None] = mapped_column(String)
 
@@ -81,10 +93,15 @@ 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
+    # pending | sent | failed — what the batch *process* did with this item;
-    # report in US05-03 and verified in US05-04; ``sent`` only means the batch
+    # ``sent`` only means the uploader exited successfully.
-    # process 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))
     updated_at: Mapped[datetime] = mapped_column(
         DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now()
     )

photo_pipeline/services/upload_batches.py

@@ -18,10 +18,11 @@ cannot undo, so the discipline is:
   vanished ``unknown_requires_verification`` (concept §15) instead of retrying it
   blindly. Resolving that is US05-04.
 
-Per-asset upload *results* are deliberately not interpreted here — parsing the
+Per-asset upload *results* are not interpreted here: after the attempt ends the
-report is US05-03. What this story guarantees is that the report exists, is bounded,
+report is handed to :class:`~photo_pipeline.services.upload_reports.
-carries no credential, and that the batch's state is an honest description of what
+UploadReportService` (US05-03), which classifies every file. The batch ``state``
-the process did.
+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 +39,7 @@ 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.uploads import UploadService
 
 
@@ -238,6 +240,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 +418,14 @@ 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,
         "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 +435,9 @@ 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,
             }
             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())

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/story_traceability.json

@@ -109,6 +109,10 @@
     ],
     "US05-02": [
       "tests/integration/test_upload_batches.py"
+    ],
+    "US05-03": [
+      "tests/unit/test_immich_go_report.py",
+      "tests/integration/test_upload_reports.py"
     ]
   }
 }

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"