US03-01: Aggregate Album Evidence

donor_ledger.yaml

@@ -336,11 +336,16 @@ rows:
       Album = leaf folder relative to library, '(root)' at root, bare parent
       outside the library. The single album-identity rule for stats, proposals
       and folder-as-album upload. webapp.query.album_of is a copy — consolidate
-      to one implementation (see wa-album-of).
+      to one implementation (see wa-album-of). Extracted in US03-01 as
+      photo_pipeline.services.albums.album_label (parity test below); the album
+      evidence aggregator uses it. LibraryService._album_of still keeps its own
+      leaf-name grouping — fold it into this rule when the Library view is next
+      touched.
     target: photo_pipeline/services/albums.py
     tests:
       - test_pa_album::test_album_label_leaf_folder_relative
       - test_pa_album::test_webapp_album_of_mirrors_album_label
+      - test_pa_album::test_albums_service_album_label_matches_donor
     status: characterized
 
   - id: pa-config
@@ -546,9 +551,12 @@ rows:
     classification: replace
     rationale: >
       Verbatim copy of photo_analyzer.album_label (equivalence characterized);
-      consolidated into the single albums-service implementation (pa-album-label).
+      consolidated into the single albums-service implementation (pa-album-label),
+      now photo_pipeline.services.albums.album_label (US03-01, parity test below).
     target: photo_pipeline/services/albums.py
-    tests: [test_pa_album::test_webapp_album_of_mirrors_album_label]
+    tests:
+      - test_pa_album::test_webapp_album_of_mirrors_album_label
+      - test_pa_album::test_albums_service_album_label_matches_donor
     status: characterized
 
   - id: wa-allowlist

photo_pipeline/services/albums.py

@@ -0,0 +1,265 @@
+"""AlbumService — aggregate per-folder evidence for album proposals (US03-01).
+
+Read-only. For each leaf folder it summarises the *already stored* analysis and
+safety evidence of the canonical, active assets living there, so a later stage can
+name the folder without resending images or exposing raw model responses. It writes
+nothing and generates no names (that is US03-02/03).
+
+Folder identity is the full parent path of an asset's ``current_path`` — not just
+the leaf name — so two folders that share a leaf never merge. Aggregation is
+deterministic (everything sorted, no clock/random) and each folder's evidence
+carries a content ``version`` derived only from the inputs that could change it, so
+any relevant asset change (new analysis, decision change, move, or becoming a
+duplicate) invalidates exactly that folder's version and nothing else.
+
+Documented evidence rules (acceptance criterion 2):
+
+- **Duplicate (non-canonical) assets** — excluded entirely; the canonical carries
+  the evidence.
+- **Missing / archived assets** — excluded: only ``availability_state == 'active'``
+  assets with a ``current_path`` belong to an active folder.
+- **NSFW assets** — counted in the folder total and reported as ``nsfw_count``, but
+  contribute NO analysis-derived evidence (tags, description, setting, location,
+  year, people). They are never analysed, and their content must not shape a name
+  beyond mere presence.
+- **Undecided / deferred / not-yet-analysed assets** — counted in the folder total
+  and reported as ``pending_count``; they contribute no evidence until confirmed
+  SFW and analysed.
+- **Only confirmed-SFW analysed assets** shape the descriptive evidence.
+- **Missing fields** (null/empty) simply do not contribute.
+- **Conflicting fields** (e.g. several ``approx_year`` values) are never silently
+  collapsed: the full sorted distribution is returned, plus a ``dominant_year`` that
+  is set only when one value strictly outnumbers the rest; ties set it to ``None``
+  and raise ``year_conflict``.
+
+Reuses ``_tags`` from :mod:`photo_pipeline.services.library` (the shared JSON-tag
+decode). ponytail: neighbouring-folder names/date ranges and user naming
+conventions (concept §7 inputs) are proposal-generation concerns — add them in
+US03-03 when a name is actually produced.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+from collections import Counter
+from pathlib import Path
+
+from sqlalchemy import select
+from sqlalchemy.orm import sessionmaker
+
+from photo_pipeline.models import AnalysisResult, Asset, SafetyReview
+from photo_pipeline.services.library import _tags
+
+SFW = "sfw"
+NSFW = "nsfw"
+ANALYZED = "analyzed"
+# Cap the descriptions list so a huge folder can't produce an unbounded payload;
+# tags/facets are already bounded by their distinct-value count.
+MAX_DESCRIPTIONS = 50
+
+
+def album_label(path: str, library_roots: tuple[Path, ...] = ()) -> str:
+    """The canonical album identity: the leaf folder directly containing ``path``,
+    shown as a path relative to its library root so nested albums stay distinct
+    (``Urlaub/Rom`` vs ``Urlaub/Venedig``); ``(root)`` for a photo directly in a
+    root, and the bare parent name for a photo outside every configured root.
+
+    Ported from ``photo_analyzer.album_label`` (donor_ledger.yaml: pa-album-label);
+    ``webapp/query.album_of`` was a verbatim copy (wa-album-of). This is the single
+    implementation for stats, proposals, and folder-as-album upload.
+    """
+    parent = Path(path).parent
+    for root in library_roots:
+        try:
+            rel = parent.relative_to(root)
+            return "(root)" if str(rel) == "." else str(rel)
+        except ValueError:
+            continue
+    return parent.name or str(parent)
+
+
+def _counts(counter: Counter) -> list[dict]:
+    """Deterministic ``[{value, count}]``, most frequent first, ties broken by value."""
+    return [
+        {"value": value, "count": count}
+        for value, count in sorted(counter.items(), key=lambda kv: (-kv[1], str(kv[0])))
+    ]
+
+
+def _dominant(counter: Counter):
+    """The single strictly-most-frequent value, or ``(None, conflict)``. ``conflict``
+    is true when more than one distinct value is present and no strict winner exists."""
+    if not counter:
+        return None, False
+    ranked = counter.most_common()
+    if len(ranked) == 1:
+        return ranked[0][0], False
+    top_value, top_count = ranked[0]
+    if top_count > ranked[1][1]:
+        return top_value, False
+    return None, True
+
+
+class AlbumService:
+    def __init__(
+        self, session_factory: sessionmaker, *, library_roots: tuple[Path, ...] = ()
+    ) -> None:
+        self._session_factory = session_factory
+        self._library_roots = tuple(Path(root) for root in library_roots)
+
+    def _album_of(self, path: str) -> str:
+        return album_label(path, self._library_roots)
+
+    def _latest_decisions(self, session) -> dict[str, str | None]:
+        """asset_id -> latest safety decision (``sfw``/``nsfw``/``deferred``/None)."""
+        latest: dict[str, str | None] = {}
+        for review in session.scalars(select(SafetyReview).order_by(SafetyReview.created_at)):
+            latest[review.asset_id] = review.decision
+        return latest
+
+    def folder_count(self) -> int:
+        """Number of distinct active-canonical folders (for paging callers)."""
+        with self._session_factory() as session:
+            paths = session.scalars(
+                select(Asset.current_path).where(
+                    Asset.canonical_asset_id.is_(None),
+                    Asset.availability_state == "active",
+                    Asset.current_path.is_not(None),
+                )
+            )
+            return len({self._album_of(p) for p in paths})
+
+    def aggregate_evidence(self, *, offset: int = 0, limit: int | None = None) -> dict:
+        """Return a page of per-folder evidence, folders sorted by path.
+
+        ``{folders: [...], total: int, offset, limit}``. ``limit=None`` returns all.
+        """
+        with self._session_factory() as session:
+            decisions = self._latest_decisions(session)
+            rows = list(
+                session.execute(
+                    select(Asset.id, Asset.current_path, AnalysisResult)
+                    .outerjoin(AnalysisResult, AnalysisResult.asset_id == Asset.id)
+                    .where(
+                        Asset.canonical_asset_id.is_(None),
+                        Asset.availability_state == "active",
+                        Asset.current_path.is_not(None),
+                    )
+                    .order_by(Asset.current_path, Asset.id)
+                )
+            )
+
+        folders: dict[str, list] = {}
+        for asset_id, path, result in rows:
+            folders.setdefault(self._album_of(path), []).append((asset_id, path, result))
+
+        ordered = sorted(folders)
+        total = len(ordered)
+        page = ordered[offset:] if limit is None else ordered[offset : offset + limit]
+        evidence = [self._folder_evidence(folder, folders[folder], decisions) for folder in page]
+        return {"folders": evidence, "total": total, "offset": offset, "limit": limit}
+
+    def _folder_evidence(self, folder: str, members: list, decisions: dict) -> dict:
+        descriptions: list[str] = []
+        tags: Counter = Counter()
+        settings: Counter = Counter()
+        time_of_day: Counter = Counter()
+        seasons: Counter = Counter()
+        locations: Counter = Counter()
+        people: Counter = Counter()
+        years: Counter = Counter()
+
+        analyzed_count = nsfw_count = pending_count = 0
+        fingerprint: list[tuple] = []
+
+        for asset_id, path, result in members:
+            decision = decisions.get(asset_id)
+            status = result.status if result is not None else None
+            # Fingerprint: every input that could change this folder's evidence.
+            fingerprint.append(
+                (
+                    asset_id,
+                    path,
+                    decision,
+                    status,
+                    result.description if result is not None else None,
+                    result.tags if result is not None else None,
+                    result.people_count if result is not None else None,
+                    result.setting if result is not None else None,
+                    result.time_of_day if result is not None else None,
+                    result.season if result is not None else None,
+                    _clean(result.location_hint) if result is not None else None,
+                    result.approx_year if result is not None else None,
+                )
+            )
+
+            if decision == NSFW:
+                nsfw_count += 1
+                continue
+            eligible = decision == SFW and status == ANALYZED and result is not None
+            if not eligible:
+                pending_count += 1
+                continue
+
+            analyzed_count += 1
+            if result.description and result.description.strip():
+                descriptions.append(result.description.strip())
+            for tag in _tags(result.tags):
+                tags[tag] += 1
+            if result.setting:
+                settings[result.setting] += 1
+            if result.time_of_day:
+                time_of_day[result.time_of_day] += 1
+            if result.season:
+                seasons[result.season] += 1
+            location = _clean(result.location_hint)
+            if location:
+                locations[location] += 1
+            if result.people_count is not None:
+                people["3+" if result.people_count >= 3 else str(result.people_count)] += 1
+            if result.approx_year is not None:
+                years[result.approx_year] += 1
+
+        dominant_year, year_conflict = _dominant(years)
+        # ``folder`` is the album label (identity): a library-relative path like
+        # ``Urlaub/Rom``, or ``(root)``. Name is its leaf; parent is the prefix.
+        parent = folder.rsplit("/", 1)[0] if "/" in folder else ""
+        name = folder.rsplit("/", 1)[-1]
+        return {
+            "album": folder,
+            "folder_name": name,
+            "parent_path": parent,
+            "asset_count": len(members),
+            "analyzed_count": analyzed_count,
+            "nsfw_count": nsfw_count,
+            "pending_count": pending_count,
+            "descriptions": sorted(descriptions)[:MAX_DESCRIPTIONS],
+            "tags": _counts(tags),
+            "settings": _counts(settings),
+            "time_of_day": _counts(time_of_day),
+            "seasons": _counts(seasons),
+            "locations": _counts(locations),
+            "people_counts": _counts(people),
+            "years": _counts(years),
+            "dominant_year": dominant_year,
+            "year_conflict": year_conflict,
+            "version": _version(fingerprint),
+        }
+
+
+def _clean(location: str | None) -> str | None:
+    """Drop empty/``null`` location hints (the model emits the string ``"null"``)."""
+    if not location:
+        return None
+    trimmed = location.strip()
+    if not trimmed or trimmed.lower() == "null":
+        return None
+    return trimmed
+
+
+def _version(fingerprint: list[tuple]) -> str:
+    """Deterministic content hash of a folder's evidence inputs. Independent of row
+    order (the inputs are sorted) so only a real change moves it."""
+    payload = json.dumps(sorted(fingerprint, key=lambda t: t[0]), ensure_ascii=False)
+    return hashlib.sha256(payload.encode("utf-8")).hexdigest()

tests/characterization/test_pa_album.py

@@ -22,6 +22,19 @@ def test_webapp_album_of_mirrors_album_label():
         assert query.album_of(p, lib) == pa.album_label(Path(p), lib), p
 
 
+def test_albums_service_album_label_matches_donor():
+    # Parity: the extracted single implementation (US03-01) matches the donor rule
+    # (donor_ledger.yaml: pa-album-label, wa-album-of). The service takes a tuple of
+    # library roots; the donor takes one Path.
+    from photo_pipeline.services.albums import album_label
+
+    lib = Path("/lib")
+    for p in ("/lib/Urlaub/Rom/a.jpg", "/lib/a.jpg", "/other/x/a.jpg", "/lib/x/y/a.jpg"):
+        assert album_label(p, (lib,)) == pa.album_label(Path(p), lib), p
+    # No configured roots → bare parent name, like the donor with library=None.
+    assert album_label("/lib/x/a.jpg", ()) == pa.album_label(Path("/lib/x/a.jpg"), None)
+
+
 def test_fit_label_left_truncates():
     assert pa.fit_label("short", 10) == "short"
     assert pa.fit_label("Urlaub/Somewhere/Rom", 8) == "…ere/Rom"

tests/integration/test_album_evidence_repo.py

@@ -0,0 +1,225 @@
+"""Repository tests for AlbumService.aggregate_evidence (US03-01): folder grouping,
+the documented exclusion/conflict rules, ordering, paging, and — the heart of the
+story — that each folder's version is stable across reads and invalidated by exactly
+the relevant asset changes, before and after those changes.
+"""
+
+import uuid
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+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
+from photo_pipeline.services.albums import AlbumService
+
+NOW = datetime(2026, 1, 1, tzinfo=timezone.utc)
+LIB = (Path("/lib"),)  # library root, so album labels are library-relative
+
+
+def _service(sf):
+    return AlbumService(sf, library_roots=LIB)
+
+
+def _factory(tmp_path):
+    (tmp_path / "data").mkdir()
+    config = Config.from_env({"PHOTO_PIPELINE_DATA_DIR": str(tmp_path / "data")})
+    run_migrations(config.database_url)
+    return create_session_factory(create_db_engine(config.database_url))
+
+
+def _asset(sf, path, *, canonical_of=None, availability="active", asset_id=None):
+    asset_id = asset_id or str(uuid.uuid4())
+    with sf() as session:
+        session.add(
+            Asset(
+                id=asset_id,
+                original_path=path,
+                current_path=path,
+                discovered_at=NOW,
+                hash_version=1,
+                availability_state=availability,
+                canonical_asset_id=canonical_of,
+            )
+        )
+        session.commit()
+    return asset_id
+
+
+def _decide(sf, asset_id, decision, *, at=NOW):
+    with sf() as session:
+        session.add(
+            SafetyReview(id=str(uuid.uuid4()), asset_id=asset_id, decision=decision, created_at=at)
+        )
+        session.commit()
+
+
+def _analyze(sf, asset_id, **fields):
+    with sf() as session:
+        row = session.get(AnalysisResult, asset_id) or AnalysisResult(asset_id=asset_id)
+        row.status = fields.pop("status", "analyzed")
+        for key, value in fields.items():
+            setattr(row, key, value)
+        session.add(row)
+        session.commit()
+
+
+def _sfw_analyzed(sf, path, **fields):
+    asset_id = _asset(sf, path)
+    _decide(sf, asset_id, "sfw")
+    _analyze(sf, asset_id, **fields)
+    return asset_id
+
+
+def _folders(sf):
+    return {f["album"]: f for f in _service(sf).aggregate_evidence()["folders"]}
+
+
+def test_groups_by_full_folder_path_and_summarises_sfw_evidence(tmp_path):
+    sf = _factory(tmp_path)
+    _sfw_analyzed(
+        sf,
+        "/lib/rome/a.jpg",
+        description="Colosseum",
+        tags='["ruins", "city"]',
+        setting="outdoor",
+        approx_year=2019,
+        people_count=2,
+        location_hint="Rome",
+    )
+    _sfw_analyzed(
+        sf,
+        "/lib/rome/b.jpg",
+        description="Forum",
+        tags='["ruins", "history"]',
+        setting="outdoor",
+        approx_year=2019,
+        people_count=4,
+        location_hint="Rome",
+    )
+
+    rome = _folders(sf)["rome"]
+    assert rome["folder_name"] == "rome" and rome["parent_path"] == ""
+    assert rome["asset_count"] == 2 and rome["analyzed_count"] == 2
+    assert rome["descriptions"] == ["Colosseum", "Forum"]  # sorted
+    assert rome["tags"][0] == {"value": "ruins", "count": 2}  # most frequent first
+    assert rome["locations"] == [{"value": "Rome", "count": 2}]
+    assert rome["people_counts"] == [{"value": "2", "count": 1}, {"value": "3+", "count": 1}]
+    assert rome["dominant_year"] == 2019 and rome["year_conflict"] is False
+
+
+def test_exclusion_rules_for_dup_nsfw_deferred_and_missing(tmp_path):
+    sf = _factory(tmp_path)
+    canonical = _sfw_analyzed(sf, "/lib/mix/keep.jpg", description="kept", tags='["a"]')
+
+    # Duplicate variant of the canonical: excluded entirely.
+    _asset(sf, "/lib/mix/dup.jpg", canonical_of=canonical)
+    # NSFW: counted, but no descriptive evidence.
+    nsfw = _asset(sf, "/lib/mix/private.jpg")
+    _decide(sf, nsfw, "nsfw")
+    # Deferred and undecided: counted as pending, no evidence.
+    deferred = _asset(sf, "/lib/mix/later.jpg")
+    _decide(sf, deferred, "deferred")
+    _asset(sf, "/lib/mix/unknown.jpg")  # no decision at all
+    # Archived: not in an active folder, excluded entirely.
+    _asset(sf, "/lib/mix/gone.jpg", availability="archived_offline")
+
+    mix = _folders(sf)["mix"]
+    # Only canonical, active assets are counted: keep + nsfw + deferred + unknown.
+    # The duplicate variant and the archived asset are excluded entirely.
+    assert mix["asset_count"] == 4
+    assert mix["nsfw_count"] == 1
+    assert mix["pending_count"] == 2  # deferred + unknown
+    assert mix["analyzed_count"] == 1
+    assert mix["descriptions"] == ["kept"] and mix["tags"] == [{"value": "a", "count": 1}]
+
+
+def test_conflicting_year_reports_distribution_and_no_dominant(tmp_path):
+    sf = _factory(tmp_path)
+    _sfw_analyzed(sf, "/lib/trip/a.jpg", approx_year=2018)
+    _sfw_analyzed(sf, "/lib/trip/b.jpg", approx_year=2019)
+
+    trip = _folders(sf)["trip"]
+    assert trip["years"] == [{"value": 2018, "count": 1}, {"value": 2019, "count": 1}]
+    assert trip["dominant_year"] is None and trip["year_conflict"] is True
+
+
+def test_folders_are_ordered_and_pageable(tmp_path):
+    sf = _factory(tmp_path)
+    for folder in ("/lib/c", "/lib/a", "/lib/b"):
+        _sfw_analyzed(sf, f"{folder}/x.jpg", description="x")
+
+    service = _service(sf)
+    assert service.folder_count() == 3
+    full = service.aggregate_evidence()
+    assert [f["album"] for f in full["folders"]] == ["a", "b", "c"]
+
+    page = service.aggregate_evidence(offset=1, limit=1)
+    assert page["total"] == 3 and page["offset"] == 1
+    assert [f["album"] for f in page["folders"]] == ["b"]
+
+
+def test_version_is_stable_across_repeated_reads(tmp_path):
+    sf = _factory(tmp_path)
+    _sfw_analyzed(sf, "/lib/rome/a.jpg", description="d", approx_year=2019)
+    first = _folders(sf)["rome"]["version"]
+    second = _folders(sf)["rome"]["version"]
+    assert first == second
+
+
+def test_version_invalidates_on_new_analysis(tmp_path):
+    sf = _factory(tmp_path)
+    _sfw_analyzed(sf, "/lib/rome/a.jpg", description="d")
+    before = _folders(sf)["rome"]["version"]
+
+    added = _asset(sf, "/lib/rome/b.jpg")
+    _decide(sf, added, "sfw")
+    _analyze(sf, added, description="new")
+    after = _folders(sf)["rome"]
+    assert after["version"] != before and after["analyzed_count"] == 2
+
+
+def test_version_invalidates_when_a_decision_changes_to_nsfw(tmp_path):
+    sf = _factory(tmp_path)
+    asset = _sfw_analyzed(sf, "/lib/rome/a.jpg", description="d")
+    before = _folders(sf)["rome"]["version"]
+
+    _decide(sf, asset, "nsfw", at=NOW + timedelta(hours=1))  # newer decision wins
+    after = _folders(sf)["rome"]
+    assert after["version"] != before
+    assert after["nsfw_count"] == 1 and after["analyzed_count"] == 0
+
+
+def test_moving_an_asset_invalidates_both_folders_and_not_a_third(tmp_path):
+    sf = _factory(tmp_path)
+    mover = _sfw_analyzed(sf, "/lib/src/a.jpg", description="d")
+    _sfw_analyzed(sf, "/lib/dst/keep.jpg", description="k")
+    _sfw_analyzed(sf, "/lib/other/z.jpg", description="z")
+    before = _folders(sf)
+    other_before = before["other"]["version"]
+
+    with sf() as session:
+        asset = session.get(Asset, mover)
+        asset.current_path = "/lib/dst/a.jpg"
+        session.commit()
+
+    after = _folders(sf)
+    assert "src" not in after  # source folder is now empty
+    assert after["dst"]["asset_count"] == 2
+    assert after["dst"]["version"] != before["dst"]["version"]
+    assert after["other"]["version"] == other_before  # untouched folder is stable
+
+
+def test_version_invalidates_when_an_asset_becomes_a_duplicate(tmp_path):
+    sf = _factory(tmp_path)
+    canonical = _sfw_analyzed(sf, "/lib/rome/a.jpg", description="a")
+    variant = _sfw_analyzed(sf, "/lib/rome/b.jpg", description="b")
+    before = _folders(sf)["rome"]["version"]
+
+    with sf() as session:
+        session.get(Asset, variant).canonical_asset_id = canonical
+        session.commit()
+
+    after = _folders(sf)["rome"]
+    assert after["version"] != before
+    assert after["asset_count"] == 1  # the variant no longer contributes

tests/story_traceability.json

@@ -68,6 +68,10 @@
     "US02-07": [
       "tests/e2e/test_phase_b_pipeline.py",
       "tests/e2e/test_analysis_browser.py"
+    ],
+    "US03-01": [
+      "tests/unit/test_album_evidence.py",
+      "tests/integration/test_album_evidence_repo.py"
     ]
   }
 }

tests/unit/test_album_evidence.py

@@ -0,0 +1,117 @@
+"""Unit + property tests for the album-evidence aggregation helpers (US03-01):
+conflict/dominant rules, deterministic ordering, location cleaning, and the content
+version's stability and sensitivity.
+"""
+
+import random
+from collections import Counter
+from pathlib import Path
+
+from photo_pipeline.services.albums import (
+    _clean,
+    _counts,
+    _dominant,
+    _version,
+    album_label,
+)
+
+
+def test_album_label_is_library_relative_and_keeps_nested_albums_distinct():
+    lib = (Path("/lib"),)
+    assert album_label("/lib/Urlaub/Rom/a.jpg", lib) == "Urlaub/Rom"
+    assert album_label("/lib/Urlaub/Venedig/a.jpg", lib) == "Urlaub/Venedig"
+    assert album_label("/lib/a.jpg", lib) == "(root)"  # directly in the root
+    assert album_label("/elsewhere/x/a.jpg", lib) == "x"  # outside → bare parent
+    assert album_label("/lib/x/a.jpg", ()) == "x"  # no configured root → bare parent
+    # Donor parity lives in tests/characterization/test_pa_album.py.
+
+
+def test_clean_drops_empty_and_literal_null_locations():
+    assert _clean(None) is None
+    assert _clean("") is None
+    assert _clean("   ") is None
+    assert _clean("null") is None
+    assert _clean("NULL") is None
+    assert _clean("  Rome  ") == "Rome"
+
+
+def test_counts_order_by_frequency_then_value():
+    counts = _counts(Counter({"beach": 1, "sunset": 3, "sea": 1}))
+    assert counts == [
+        {"value": "sunset", "count": 3},
+        {"value": "beach", "count": 1},  # tie broken alphabetically
+        {"value": "sea", "count": 1},
+    ]
+
+
+def test_dominant_requires_a_strict_winner():
+    assert _dominant(Counter()) == (None, False)
+    assert _dominant(Counter({2019: 3})) == (2019, False)
+    assert _dominant(Counter({2019: 3, 2020: 1})) == (2019, False)
+    # A tie is a conflict: no dominant value, flagged.
+    assert _dominant(Counter({2019: 2, 2020: 2})) == (None, True)
+
+
+def test_version_is_stable_and_row_order_independent():
+    a = [("id1", "/f/a.jpg", "sfw", "analyzed", "d", None, 1, None, None, None, None, 2019)]
+    b = list(a)
+    two = a + [("id2", "/f/b.jpg", "sfw", "analyzed", "e", None, 2, None, None, None, None, 2019)]
+    assert _version(a) == _version(b)
+    # Reversed input order must not change the version (inputs are sorted).
+    assert _version(two) == _version(list(reversed(two)))
+
+
+def test_version_changes_when_any_input_field_changes():
+    base = (
+        "id1",
+        "/f/a.jpg",
+        "sfw",
+        "analyzed",
+        "desc",
+        None,
+        1,
+        "outdoor",
+        None,
+        None,
+        None,
+        2019,
+    )
+    baseline = _version([base])
+    # Flip each field in turn; every change must move the hash.
+    for index in range(len(base)):
+        changed = list(base)
+        current = changed[index]
+        changed[index] = 99 if isinstance(current, int) else "CHANGED"
+        assert _version([tuple(changed)]) != baseline, f"field {index} did not affect the version"
+
+
+def test_property_dominant_matches_a_brute_force_definition():
+    rng = random.Random(1234)
+    for _ in range(500):
+        counter = Counter()
+        for _ in range(rng.randint(0, 6)):
+            counter[rng.randint(2000, 2005)] += 1
+        value, conflict = _dominant(counter)
+        if not counter:
+            assert value is None and conflict is False
+            continue
+        top = max(counter.values())
+        winners = [k for k, v in counter.items() if v == top]
+        if len(winners) == 1:
+            assert value == winners[0] and conflict is False
+        else:
+            assert value is None and conflict is True
+
+
+def test_property_counts_is_a_permutation_sorted_by_frequency():
+    rng = random.Random(99)
+    for _ in range(300):
+        counter = Counter()
+        for _ in range(rng.randint(0, 20)):
+            counter[rng.choice("abcde")] += 1
+        counts = _counts(counter)
+        # Same multiset of (value, count) pairs as the counter.
+        assert {(c["value"], c["count"]) for c in counts} == set(counter.items())
+        # Non-increasing frequency.
+        freqs = [c["count"] for c in counts]
+        assert freqs == sorted(freqs, reverse=True)