migrations/versions/0002_duplicate_engine.py

@@ -0,0 +1,99 @@
+"""Duplicate engine: canonical link, phash version, clusters, members, negative links.
+
+Revision ID: 0002_duplicate_engine
+Revises: 0001_initial_identity
+Create Date: 2026-07-15
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = "0002_duplicate_engine"
+down_revision = "0001_initial_identity"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    # Plain ADD COLUMN: SQLite cannot ALTER-ADD a foreign key, and the ORM model
+    # keeps the ForeignKey for relationship metadata. canonical_asset_id is a soft
+    # self-link validated in the service (no cycles), not DB-enforced.
+    op.add_column("assets", sa.Column("phash_version", sa.Integer(), nullable=True))
+    op.add_column("assets", sa.Column("canonical_asset_id", sa.String(), nullable=True))
+    op.create_index("ix_assets_canonical_asset_id", "assets", ["canonical_asset_id"])
+
+    op.create_table(
+        "duplicate_clusters",
+        sa.Column("id", sa.String(), primary_key=True),
+        sa.Column("method", sa.String(), nullable=False),
+        sa.Column("confidence", sa.String(), nullable=False),
+        sa.Column("state", sa.String(), nullable=False),
+        sa.Column("decision", sa.String(), nullable=True),
+        sa.Column(
+            "canonical_asset_id", sa.String(), sa.ForeignKey("assets.id"), nullable=True
+        ),
+        sa.Column("version", sa.Integer(), nullable=False, server_default="1"),
+        sa.Column(
+            "created_at",
+            sa.DateTime(timezone=True),
+            nullable=False,
+            server_default=sa.text("CURRENT_TIMESTAMP"),
+        ),
+        sa.Column(
+            "updated_at",
+            sa.DateTime(timezone=True),
+            nullable=False,
+            server_default=sa.text("CURRENT_TIMESTAMP"),
+        ),
+        sa.Column("decided_at", sa.DateTime(timezone=True), nullable=True),
+    )
+    op.create_index("ix_duplicate_clusters_state", "duplicate_clusters", ["state"])
+
+    op.create_table(
+        "duplicate_members",
+        sa.Column(
+            "cluster_id",
+            sa.String(),
+            sa.ForeignKey("duplicate_clusters.id"),
+            primary_key=True,
+        ),
+        sa.Column(
+            "asset_id", sa.String(), sa.ForeignKey("assets.id"), primary_key=True
+        ),
+        sa.Column("role", sa.String(), nullable=False, server_default="member"),
+        sa.Column("distance", sa.Integer(), nullable=True),
+        sa.Column("evidence", sa.String(), nullable=True),
+        sa.Column(
+            "added_at",
+            sa.DateTime(timezone=True),
+            nullable=False,
+            server_default=sa.text("CURRENT_TIMESTAMP"),
+        ),
+    )
+    op.create_index("ix_duplicate_members_asset_id", "duplicate_members", ["asset_id"])
+
+    op.create_table(
+        "duplicate_negative_links",
+        sa.Column(
+            "asset_a", sa.String(), sa.ForeignKey("assets.id"), primary_key=True
+        ),
+        sa.Column(
+            "asset_b", sa.String(), sa.ForeignKey("assets.id"), primary_key=True
+        ),
+        sa.Column("reason", sa.String(), nullable=True),
+        sa.Column(
+            "created_at",
+            sa.DateTime(timezone=True),
+            nullable=False,
+            server_default=sa.text("CURRENT_TIMESTAMP"),
+        ),
+    )
+
+
+def downgrade() -> None:
+    op.drop_table("duplicate_negative_links")
+    op.drop_table("duplicate_members")
+    op.drop_table("duplicate_clusters")
+    op.drop_index("ix_assets_canonical_asset_id", "assets")
+    op.drop_column("assets", "canonical_asset_id")
+    op.drop_column("assets", "phash_version")

photo_pipeline/models/__init__.py

@@ -5,5 +5,16 @@ Alembic environment relies on.
 """
 
 from photo_pipeline.models.assets import Asset, AssetPath
+from photo_pipeline.models.duplicates import (
+    DuplicateCluster,
+    DuplicateMember,
+    DuplicateNegativeLink,
+)
 
-__all__ = ["Asset", "AssetPath"]
+__all__ = [
+    "Asset",
+    "AssetPath",
+    "DuplicateCluster",
+    "DuplicateMember",
+    "DuplicateNegativeLink",
+]

photo_pipeline/models/assets.py

@@ -30,11 +30,14 @@ class Asset(Base):
     pixel_sha256: Mapped[str | None] = mapped_column(String)
     phash: Mapped[str | None] = mapped_column(String)
     hash_version: Mapped[int] = mapped_column(Integer, nullable=False, default=1)
+    phash_version: Mapped[int | None] = mapped_column(Integer)
     byte_size: Mapped[int | None] = mapped_column(Integer)
 
     discovered_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
     missing_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
     availability_state: Mapped[str] = mapped_column(String, nullable=False, default="active")
+    # Duplicate canonical link: NULL when the asset is itself canonical or undecided.
+    canonical_asset_id: Mapped[str | None] = mapped_column(ForeignKey("assets.id"))
 
     state_version: Mapped[int] = mapped_column(Integer, nullable=False, default=1)
     created_at: Mapped[datetime] = mapped_column(

photo_pipeline/models/duplicates.py

@@ -0,0 +1,62 @@
+"""Duplicate clusters, their members, and not-duplicate negative links.
+
+A cluster is a reviewable grouping produced by a detection ``method`` (exact,
+pixel, or perceptual) at a ``confidence`` band. Exact/pixel clusters can be
+decided automatically; perceptual clusters stay open for human review. Every
+member carries the evidence behind its inclusion. Negative links record
+``not_duplicate`` decisions so rescans do not re-suggest a rejected pair.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime
+
+from sqlalchemy import DateTime, ForeignKey, Integer, String, func
+from sqlalchemy.orm import Mapped, mapped_column
+
+from photo_pipeline.db import Base
+
+
+class DuplicateCluster(Base):
+    __tablename__ = "duplicate_clusters"
+
+    id: Mapped[str] = mapped_column(String, primary_key=True)
+    method: Mapped[str] = mapped_column(String, nullable=False)  # exact|pixel|perceptual
+    confidence: Mapped[str] = mapped_column(String, nullable=False)  # band name
+    state: Mapped[str] = mapped_column(String, nullable=False)  # open|decided|dismissed|reopened
+    decision: Mapped[str | None] = mapped_column(String)  # canonical|not_duplicate|deferred
+    canonical_asset_id: Mapped[str | None] = mapped_column(ForeignKey("assets.id"))
+    version: Mapped[int] = mapped_column(Integer, nullable=False, default=1)
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, server_default=func.now()
+    )
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now()
+    )
+    decided_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+
+
+class DuplicateMember(Base):
+    __tablename__ = "duplicate_members"
+
+    cluster_id: Mapped[str] = mapped_column(
+        ForeignKey("duplicate_clusters.id"), primary_key=True
+    )
+    asset_id: Mapped[str] = mapped_column(ForeignKey("assets.id"), primary_key=True)
+    role: Mapped[str] = mapped_column(String, nullable=False, default="member")  # member|canonical|variant
+    distance: Mapped[int | None] = mapped_column(Integer)  # phash distance to representative
+    evidence: Mapped[str | None] = mapped_column(String)  # JSON blob explaining inclusion
+    added_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, server_default=func.now()
+    )
+
+
+class DuplicateNegativeLink(Base):
+    __tablename__ = "duplicate_negative_links"
+
+    asset_a: Mapped[str] = mapped_column(ForeignKey("assets.id"), primary_key=True)
+    asset_b: Mapped[str] = mapped_column(ForeignKey("assets.id"), primary_key=True)
+    reason: Mapped[str | None] = mapped_column(String)
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, server_default=func.now()
+    )

photo_pipeline/services/duplicates.py

@@ -0,0 +1,568 @@
+"""DuplicateService — explainable clusters and reversible decisions.
+
+Detection runs in two categories:
+
+- **Content-identical** clusters join assets that share exact bytes (``sha256``) or
+  normalized pixels (``pixel_sha256``). These are high confidence and are decided
+  automatically (a recommended canonical, the rest variants) — unless a member pair
+  carries a ``not_duplicate`` negative link, which is a contradiction that reopens
+  the cluster for review instead of silently overriding the user.
+- **Perceptual** clusters join assets whose perceptual hashes are within a distance
+  band (NEAR/SIMILAR). These are review candidates: never decided automatically, and
+  negative-linked pairs are suppressed so a rejected pair is not re-suggested.
+
+Decisions (``canonical`` / ``not_duplicate`` / ``deferred``) persist with evidence,
+use optimistic version checks, are reversible, and can never form a canonical cycle.
+A new content-identical member of an already-decided cluster inherits the established
+canonical relationship rather than reopening it.
+
+Extracts photo_analyzer.cluster_duplicates/mark_duplicates (union-find over the
+Hamming graph); path-keyed identity and automatic largest-file selection are
+replaced by stable asset IDs and reviewable decisions (donor_ledger.yaml: pa-dupes).
+"""
+
+from __future__ import annotations
+
+import json
+import uuid
+from collections import Counter
+from dataclasses import dataclass, field
+from datetime import datetime, timezone
+from enum import Enum
+
+from sqlalchemy import select
+from sqlalchemy.orm import sessionmaker
+
+from photo_pipeline.models import (
+    Asset,
+    DuplicateCluster,
+    DuplicateMember,
+    DuplicateNegativeLink,
+)
+from photo_pipeline.services import hashing
+
+NEAR_MAX = 5
+SIMILAR_MAX = 10
+
+
+class Method(str, Enum):
+    EXACT = "exact"
+    PIXEL = "pixel"
+    PERCEPTUAL = "perceptual"
+
+
+class Confidence(str, Enum):
+    EXACT = "exact"
+    PIXEL = "pixel"
+    NEAR = "near"
+    SIMILAR = "similar"
+
+
+class ClusterState(str, Enum):
+    OPEN = "open"
+    DECIDED = "decided"
+    DISMISSED = "dismissed"
+    REOPENED = "reopened"
+    DEFERRED = "deferred"
+
+
+class Decision(str, Enum):
+    CANONICAL = "canonical"
+    NOT_DUPLICATE = "not_duplicate"
+    DEFERRED = "deferred"
+
+
+class Role(str, Enum):
+    MEMBER = "member"
+    CANONICAL = "canonical"
+    VARIANT = "variant"
+
+
+class DuplicateError(RuntimeError):
+    """Invalid decision (unknown cluster, non-member canonical, would cycle)."""
+
+
+class ConflictError(RuntimeError):
+    """Optimistic version check failed; the cluster changed since it was read."""
+
+
+@dataclass
+class DetectionReport:
+    clusters: list[dict] = field(default_factory=list)
+
+    @property
+    def counts(self) -> dict[str, int]:
+        return dict(Counter(c["state"] for c in self.clusters))
+
+
+def _pair(a: str, b: str) -> tuple[str, str]:
+    return (a, b) if a <= b else (b, a)
+
+
+class _UnionFind:
+    def __init__(self, items):
+        self.parent = {x: x for x in items}
+
+    def find(self, x):
+        while self.parent[x] != x:
+            self.parent[x] = self.parent[self.parent[x]]
+            x = self.parent[x]
+        return x
+
+    def union(self, a, b):
+        ra, rb = self.find(a), self.find(b)
+        if ra != rb:
+            self.parent[rb] = ra
+
+    def groups(self) -> list[list[str]]:
+        out: dict[str, list[str]] = {}
+        for x in self.parent:
+            out.setdefault(self.find(x), []).append(x)
+        return [sorted(g) for g in out.values()]
+
+
+class DuplicateService:
+    def __init__(self, session_factory: sessionmaker) -> None:
+        self._session_factory = session_factory
+
+    # ── perceptual hash backfill ───────────────────────────────────────────
+    def ensure_phashes(self) -> int:
+        updated = 0
+        with self._session_factory() as session:
+            assets = session.execute(
+                select(Asset).where(
+                    Asset.availability_state == "active",
+                    Asset.current_path.isnot(None),
+                )
+            ).scalars()
+            for asset in assets:
+                if asset.phash is not None and asset.phash_version == hashing.PHASH_VERSION:
+                    continue
+                value = hashing.safe_phash(asset.current_path)
+                if value is not None:
+                    asset.phash = value
+                    asset.phash_version = hashing.PHASH_VERSION
+                    updated += 1
+            session.commit()
+        return updated
+
+    # ── detection ──────────────────────────────────────────────────────────
+    def detect(self) -> DetectionReport:
+        self.ensure_phashes()
+        now = datetime.now(timezone.utc)
+        report = DetectionReport()
+        with self._session_factory() as session:
+            assets = list(
+                session.execute(
+                    select(Asset).where(
+                        Asset.availability_state == "active",
+                        Asset.current_path.isnot(None),
+                    )
+                ).scalars()
+            )
+            by_id = {a.id: a for a in assets}
+            negatives = {
+                _pair(link.asset_a, link.asset_b)
+                for link in session.execute(select(DuplicateNegativeLink)).scalars()
+            }
+            clusters = list(session.execute(select(DuplicateCluster)).scalars())
+            members = {
+                c.id: list(
+                    session.execute(
+                        select(DuplicateMember).where(DuplicateMember.cluster_id == c.id)
+                    ).scalars()
+                )
+                for c in clusters
+            }
+
+            content_groups, content_pairs = self._content_groups(assets)
+            perceptual_groups = self._perceptual_groups(assets, content_pairs, negatives)
+
+            touched: list[DuplicateCluster] = []
+            for method, ids, distances in content_groups:
+                touched.append(
+                    self._sync_content_cluster(
+                        session, method, ids, by_id, negatives, clusters, members, now
+                    )
+                )
+            for ids, distances in perceptual_groups:
+                touched.append(
+                    self._sync_perceptual_cluster(
+                        session, ids, distances, by_id, clusters, members, now
+                    )
+                )
+            session.commit()
+            report.clusters = [self._snapshot(session, c.id) for c in touched]
+        return report
+
+    def _content_groups(self, assets):
+        """Union-find over exact-byte and same-pixel edges. Returns groups of >=2
+        with method EXACT (all same bytes) or PIXEL, plus the set of joined pairs."""
+        uf = _UnionFind([a.id for a in assets])
+        pairs: set[tuple[str, str]] = set()
+        for key in ("current_sha256", "pixel_sha256"):
+            buckets: dict[str, list[str]] = {}
+            for a in assets:
+                value = getattr(a, key)
+                if value:
+                    buckets.setdefault(value, []).append(a.id)
+            for group in buckets.values():
+                for other in group[1:]:
+                    uf.union(group[0], other)
+                for i in range(len(group)):
+                    for j in range(i + 1, len(group)):
+                        pairs.add(_pair(group[i], group[j]))
+        by_id = {a.id: a for a in assets}
+        result = []
+        for group in uf.groups():
+            if len(group) < 2:
+                continue
+            shas = {by_id[i].current_sha256 for i in group}
+            method = Method.EXACT if len(shas) == 1 and None not in shas else Method.PIXEL
+            result.append((method, sorted(group), {}))
+        return result, pairs
+
+    def _perceptual_groups(self, assets, content_pairs, negatives):
+        """Union-find over phash distance edges (<= SIMILAR_MAX), excluding pairs
+        that are already content-identical or negatively linked."""
+        hashed = [a for a in assets if a.phash]
+        uf = _UnionFind([a.id for a in hashed])
+        distances: dict[tuple[str, str], int] = {}
+        for i in range(len(hashed)):
+            for j in range(i + 1, len(hashed)):
+                a, b = hashed[i], hashed[j]
+                pair = _pair(a.id, b.id)
+                if pair in content_pairs or pair in negatives:
+                    continue
+                dist = hashing.phash_distance(a.phash, b.phash)
+                if dist <= SIMILAR_MAX:
+                    uf.union(a.id, b.id)
+                    distances[pair] = dist
+        result = []
+        for group in uf.groups():
+            if len(group) < 2:
+                continue
+            group_dists = {
+                _pair(x, y): distances[_pair(x, y)]
+                for x in group
+                for y in group
+                if x < y and _pair(x, y) in distances
+            }
+            result.append((sorted(group), group_dists))
+        return result
+
+    # ── cluster upserts ──────────────────────────────────────────────────────
+    def _match_existing(self, ids, clusters, members, categories):
+        best = None
+        best_overlap = 0
+        for cluster in clusters:
+            if cluster.method not in categories:
+                continue
+            overlap = len(set(ids) & {m.asset_id for m in members[cluster.id]})
+            if overlap > best_overlap:
+                best, best_overlap = cluster, overlap
+        return best
+
+    def _sync_content_cluster(
+        self, session, method, ids, by_id, negatives, clusters, members, now
+    ):
+        contradiction = any(
+            _pair(ids[i], ids[j]) in negatives
+            for i in range(len(ids))
+            for j in range(i + 1, len(ids))
+        )
+        confidence = Confidence.EXACT if method == Method.EXACT else Confidence.PIXEL
+        # Also consider a prior perceptual cluster over the same assets: if a member
+        # pair became byte-identical after a not_duplicate decision, reopen that same
+        # reviewed cluster rather than spawning a parallel one.
+        existing = self._match_existing(
+            ids,
+            clusters,
+            members,
+            {Method.EXACT.value, Method.PIXEL.value, Method.PERCEPTUAL.value},
+        )
+
+        if existing is None:
+            cluster = DuplicateCluster(
+                id=str(uuid.uuid4()),
+                method=method.value,
+                confidence=confidence.value,
+                state=ClusterState.OPEN.value,
+                version=1,
+            )
+            session.add(cluster)
+            session.flush()
+            self._replace_members(session, cluster, ids, by_id, distances={})
+            if contradiction:
+                cluster.state = ClusterState.REOPENED.value
+            else:
+                self._auto_decide(session, cluster, ids, by_id, now)
+            clusters.append(cluster)
+            members[cluster.id] = list(
+                session.execute(
+                    select(DuplicateMember).where(DuplicateMember.cluster_id == cluster.id)
+                ).scalars()
+            )
+            return cluster
+
+        cluster = existing
+        cluster.method = method.value
+        cluster.confidence = confidence.value
+        prior_members = {m.asset_id for m in members[cluster.id]}
+        added = set(ids) - prior_members
+        self._replace_members(session, cluster, ids, by_id, distances={})
+
+        if contradiction:
+            cluster.state = ClusterState.REOPENED.value
+            cluster.decision = None
+            cluster.canonical_asset_id = None
+            self._clear_asset_links(session, ids)
+        elif (
+            cluster.decision == Decision.CANONICAL.value
+            and cluster.canonical_asset_id in ids
+        ):
+            # Consistent growth: new members inherit the established canonical.
+            self._apply_canonical(session, cluster, ids, cluster.canonical_asset_id)
+            cluster.state = ClusterState.DECIDED.value
+        elif cluster.state in (ClusterState.DISMISSED.value,) and added:
+            cluster.state = ClusterState.REOPENED.value
+        elif cluster.state in (ClusterState.OPEN.value, ClusterState.REOPENED.value):
+            self._auto_decide(session, cluster, ids, by_id, now)
+        cluster.version += 1
+        cluster.updated_at = now
+        members[cluster.id] = list(
+            session.execute(
+                select(DuplicateMember).where(DuplicateMember.cluster_id == cluster.id)
+            ).scalars()
+        )
+        return cluster
+
+    def _sync_perceptual_cluster(self, session, ids, distances, by_id, clusters, members, now):
+        confidence = (
+            Confidence.NEAR
+            if distances and max(distances.values()) <= NEAR_MAX
+            else Confidence.SIMILAR
+        )
+        existing = self._match_existing(ids, clusters, members, {Method.PERCEPTUAL.value})
+        if existing is None:
+            cluster = DuplicateCluster(
+                id=str(uuid.uuid4()),
+                method=Method.PERCEPTUAL.value,
+                confidence=confidence.value,
+                state=ClusterState.OPEN.value,
+                version=1,
+            )
+            session.add(cluster)
+            session.flush()
+            self._replace_members(session, cluster, ids, by_id, distances)
+            clusters.append(cluster)
+            members[cluster.id] = list(
+                session.execute(
+                    select(DuplicateMember).where(DuplicateMember.cluster_id == cluster.id)
+                ).scalars()
+            )
+            return cluster
+
+        cluster = existing
+        prior = {m.asset_id for m in members[cluster.id]}
+        added = set(ids) - prior
+        cluster.confidence = confidence.value
+        self._replace_members(session, cluster, ids, by_id, distances)
+        if added and cluster.state in (
+            ClusterState.DECIDED.value,
+            ClusterState.DISMISSED.value,
+            ClusterState.DEFERRED.value,
+        ):
+            cluster.state = ClusterState.REOPENED.value
+        cluster.version += 1
+        cluster.updated_at = now
+        members[cluster.id] = list(
+            session.execute(
+                select(DuplicateMember).where(DuplicateMember.cluster_id == cluster.id)
+            ).scalars()
+        )
+        return cluster
+
+    def _replace_members(self, session, cluster, ids, by_id, distances):
+        current = {
+            m.asset_id: m
+            for m in session.execute(
+                select(DuplicateMember).where(DuplicateMember.cluster_id == cluster.id)
+            ).scalars()
+        }
+        wanted = set(ids)
+        for asset_id, member in current.items():
+            if asset_id not in wanted:
+                session.delete(member)
+        for asset_id in ids:
+            asset = by_id[asset_id]
+            evidence = json.dumps(
+                {
+                    "sha256": asset.current_sha256,
+                    "pixel_sha256": asset.pixel_sha256,
+                    "phash": asset.phash,
+                }
+            )
+            distance = next(
+                (d for p, d in distances.items() if asset_id in p), None
+            )
+            if asset_id in current:
+                current[asset_id].evidence = evidence
+                current[asset_id].distance = distance
+            else:
+                session.add(
+                    DuplicateMember(
+                        cluster_id=cluster.id,
+                        asset_id=asset_id,
+                        role=Role.MEMBER.value,
+                        distance=distance,
+                        evidence=evidence,
+                    )
+                )
+        session.flush()
+
+    def _auto_decide(self, session, cluster, ids, by_id, now):
+        canonical = self._recommend_canonical(ids, by_id)
+        self._apply_canonical(session, cluster, ids, canonical)
+        cluster.decision = Decision.CANONICAL.value
+        cluster.canonical_asset_id = canonical
+        cluster.state = ClusterState.DECIDED.value
+        cluster.decided_at = now
+
+    @staticmethod
+    def _recommend_canonical(ids, by_id) -> str:
+        # ponytail: largest file, path as deterministic tie-break. The concept's
+        # richer policy (resolution, least recompression, metadata richness) lands
+        # with the review UI story.
+        return max(ids, key=lambda i: (by_id[i].byte_size or 0, by_id[i].current_path or ""))
+
+    def _apply_canonical(self, session, cluster, ids, canonical_id):
+        for member in session.execute(
+            select(DuplicateMember).where(DuplicateMember.cluster_id == cluster.id)
+        ).scalars():
+            asset = session.get(Asset, member.asset_id)
+            if member.asset_id == canonical_id:
+                member.role = Role.CANONICAL.value
+                asset.canonical_asset_id = None
+            else:
+                member.role = Role.VARIANT.value
+                asset.canonical_asset_id = canonical_id
+        self._assert_acyclic(session, ids)
+
+    @staticmethod
+    def _clear_asset_links(session, ids):
+        for asset_id in ids:
+            session.get(Asset, asset_id).canonical_asset_id = None
+
+    @staticmethod
+    def _assert_acyclic(session, ids):
+        for start in ids:
+            seen = set()
+            cur = start
+            while cur is not None:
+                if cur in seen:
+                    raise DuplicateError(f"canonical link cycle at {cur}")
+                seen.add(cur)
+                asset = session.get(Asset, cur)
+                cur = asset.canonical_asset_id if asset else None
+
+    # ── decisions ────────────────────────────────────────────────────────────
+    def decide(
+        self,
+        cluster_id: str,
+        decision: Decision | str,
+        *,
+        canonical_asset_id: str | None = None,
+        expected_version: int | None = None,
+        reason: str | None = None,
+    ) -> dict:
+        decision = Decision(decision)
+        now = datetime.now(timezone.utc)
+        with self._session_factory() as session:
+            cluster = session.get(DuplicateCluster, cluster_id)
+            if cluster is None:
+                raise DuplicateError(f"unknown cluster {cluster_id}")
+            if expected_version is not None and cluster.version != expected_version:
+                raise ConflictError(
+                    f"cluster {cluster_id} is at version {cluster.version}, "
+                    f"expected {expected_version}"
+                )
+            member_ids = [
+                m.asset_id
+                for m in session.execute(
+                    select(DuplicateMember).where(DuplicateMember.cluster_id == cluster_id)
+                ).scalars()
+            ]
+            self._reset_members(session, cluster, member_ids)
+            self._remove_negative_links(session, member_ids)
+
+            if decision == Decision.CANONICAL:
+                if canonical_asset_id not in member_ids:
+                    raise DuplicateError("canonical_asset_id must be a cluster member")
+                self._apply_canonical(session, cluster, member_ids, canonical_asset_id)
+                cluster.decision = Decision.CANONICAL.value
+                cluster.canonical_asset_id = canonical_asset_id
+                cluster.state = ClusterState.DECIDED.value
+            elif decision == Decision.NOT_DUPLICATE:
+                self._add_negative_links(session, member_ids, reason)
+                cluster.decision = Decision.NOT_DUPLICATE.value
+                cluster.canonical_asset_id = None
+                cluster.state = ClusterState.DISMISSED.value
+            else:  # DEFERRED
+                cluster.decision = Decision.DEFERRED.value
+                cluster.canonical_asset_id = None
+                cluster.state = ClusterState.DEFERRED.value
+
+            cluster.decided_at = now
+            cluster.updated_at = now
+            cluster.version += 1
+            session.commit()
+            return self._snapshot(session, cluster_id)
+
+    def _reset_members(self, session, cluster, member_ids):
+        for member in session.execute(
+            select(DuplicateMember).where(DuplicateMember.cluster_id == cluster.id)
+        ).scalars():
+            member.role = Role.MEMBER.value
+        for asset_id in member_ids:
+            session.get(Asset, asset_id).canonical_asset_id = None
+
+    @staticmethod
+    def _add_negative_links(session, member_ids, reason):
+        existing = {
+            _pair(link.asset_a, link.asset_b)
+            for link in session.execute(select(DuplicateNegativeLink)).scalars()
+        }
+        for i in range(len(member_ids)):
+            for j in range(i + 1, len(member_ids)):
+                a, b = _pair(member_ids[i], member_ids[j])
+                if (a, b) not in existing:
+                    session.add(
+                        DuplicateNegativeLink(asset_a=a, asset_b=b, reason=reason)
+                    )
+
+    @staticmethod
+    def _remove_negative_links(session, member_ids):
+        member_set = set(member_ids)
+        for link in session.execute(select(DuplicateNegativeLink)).scalars():
+            if link.asset_a in member_set and link.asset_b in member_set:
+                session.delete(link)
+
+    def _snapshot(self, session, cluster_id) -> dict:
+        cluster = session.get(DuplicateCluster, cluster_id)
+        members = session.execute(
+            select(DuplicateMember).where(DuplicateMember.cluster_id == cluster_id)
+        ).scalars()
+        return {
+            "id": cluster.id,
+            "method": cluster.method,
+            "confidence": cluster.confidence,
+            "state": cluster.state,
+            "decision": cluster.decision,
+            "canonical_asset_id": cluster.canonical_asset_id,
+            "version": cluster.version,
+            "members": sorted(
+                ({"asset_id": m.asset_id, "role": m.role, "distance": m.distance} for m in members),
+                key=lambda m: m["asset_id"],
+            ),
+        }

photo_pipeline/services/hashing.py

@@ -19,6 +19,7 @@ from pathlib import Path
 from PIL import Image, ImageOps
 
 PIXEL_HASH_VERSION = 1
+PHASH_VERSION = 1
 _CHUNK = 1 << 20
 
 
@@ -45,3 +46,39 @@ def safe_pixel_sha256(path: Path | str) -> str | None:
         return pixel_sha256(path)
     except Exception:
         return None
+
+
+def phash(path: Path | str) -> str:
+    """DCT perceptual hash → 16 hex chars (64 bits).
+
+    Standard imagehash.phash recipe (extracted verbatim from
+    photo_analyzer._phash_image): grayscale → 32x32 → 2D DCT-II → keep the top-left
+    8x8 low-frequency block → bit = coefficient > median. Resilient to resize and
+    recompression; used only as evidence for review, never for automatic exclusion.
+    """
+    import numpy as np
+    from scipy.fftpack import dct
+
+    with Image.open(path) as image:
+        small = image.convert("L").resize((32, 32), Image.LANCZOS)
+        matrix = np.asarray(small, dtype=np.float64)
+    transformed = dct(dct(matrix, axis=0), axis=1)
+    low = transformed[:8, :8]
+    bits = (low > np.median(low)).flatten()
+    value = 0
+    for bit in bits:
+        value = (value << 1) | int(bit)
+    return f"{value:016x}"
+
+
+def safe_phash(path: Path | str) -> str | None:
+    """``phash`` but returns None for undecodable images instead of raising."""
+    try:
+        return phash(path)
+    except Exception:
+        return None
+
+
+def phash_distance(a_hex: str, b_hex: str) -> int:
+    """Hamming distance between two 64-bit hex perceptual hashes."""
+    return bin(int(a_hex, 16) ^ int(b_hex, 16)).count("1")

tests/integration/test_duplicate_engine.py

@@ -0,0 +1,291 @@
+"""Duplicate engine: banded clustering, reversible decisions, negative links,
+reopening on contradiction, no canonical cycles, and durability across restart.
+
+Fixtures are inlined (not in a sibling conftest.py) to avoid shadowing the
+characterization suite's bare ``conftest`` import under pytest's prepend mode.
+"""
+
+import shutil
+
+import numpy as np
+import pytest
+from PIL import Image
+from sqlalchemy import select
+
+from photo_pipeline.db import create_db_engine, create_session_factory, run_migrations
+from photo_pipeline.models import Asset, DuplicateNegativeLink
+from photo_pipeline.services.duplicates import (
+    ClusterState,
+    ConflictError,
+    Decision,
+    DuplicateError,
+    DuplicateService,
+    Method,
+)
+from photo_pipeline.services.inventory import InventoryService
+
+
+@pytest.fixture
+def db_url(tmp_path):
+    url = f"sqlite:///{tmp_path / 'dup.db'}"
+    run_migrations(url)
+    return url
+
+
+@pytest.fixture
+def make_factory():
+    engines = []
+
+    def _make(url):
+        engine = create_db_engine(url)
+        engines.append(engine)
+        return create_session_factory(engine)
+
+    yield _make
+    for engine in engines:
+        engine.dispose()
+
+
+@pytest.fixture
+def factory(db_url, make_factory):
+    return make_factory(db_url)
+
+
+@pytest.fixture
+def inventory(factory):
+    return InventoryService(factory)
+
+
+@pytest.fixture
+def duplicates(factory):
+    return DuplicateService(factory)
+
+
+def structured(path, seed, size=(256, 192)):
+    path.parent.mkdir(parents=True, exist_ok=True)
+    rng = np.random.default_rng(seed)
+    w, h = size
+    base = np.zeros((h, w, 3), dtype=np.uint8)
+    for _ in range(6):
+        x0 = int(rng.integers(0, w - 60))
+        y0 = int(rng.integers(0, h - 60))
+        base[y0 : y0 + 60, x0 : x0 + 60] = rng.integers(0, 256, 3)
+    grad = np.linspace(0, 120, w, dtype=np.uint8)
+    base[:, :, 0] = np.clip(base[:, :, 0].astype(int) + grad[None, :], 0, 255)
+    Image.fromarray(base).save(path, quality=95)
+    return path
+
+
+def resized_copy(src, dst, scale=0.5):
+    with Image.open(src) as image:
+        image.resize(
+            (int(image.width * scale), int(image.height * scale)), Image.LANCZOS
+        ).save(dst, quality=95)
+    return dst
+
+
+def only(clusters, method):
+    return [c for c in clusters if c["method"] == method]
+
+
+def test_exact_copies_form_auto_decided_cluster(tmp_path, inventory, duplicates, factory):
+    lib = tmp_path / "lib"
+    a = structured(lib / "a.jpg", 1)
+    shutil.copy2(a, lib / "a_copy.jpg")
+    inventory.scan(lib)
+
+    clusters = duplicates.detect().clusters
+    exact = only(clusters, Method.EXACT.value)
+    assert len(exact) == 1
+    cluster = exact[0]
+    assert cluster["state"] == ClusterState.DECIDED.value
+    assert cluster["decision"] == Decision.CANONICAL.value
+    roles = {m["role"] for m in cluster["members"]}
+    assert roles == {"canonical", "variant"}
+
+    # The variant asset points at the canonical; the canonical points at nothing.
+    assets = {a.id: a for a in factory().execute(select(Asset)).scalars()}
+    canonical = cluster["canonical_asset_id"]
+    for m in cluster["members"]:
+        if m["role"] == "variant":
+            assert assets[m["asset_id"]].canonical_asset_id == canonical
+        else:
+            assert assets[m["asset_id"]].canonical_asset_id is None
+
+
+def test_same_pixels_different_bytes_form_pixel_cluster(tmp_path, inventory, duplicates):
+    lib = tmp_path / "lib"
+    a = structured(lib / "a.jpg", 1)
+    twin = lib / "a_meta.jpg"
+    shutil.copy2(a, twin)
+    with open(twin, "ab") as handle:  # trailing bytes: same pixels, different bytes
+        handle.write(b"\xff\xfe metadata")
+    inventory.scan(lib)
+
+    clusters = duplicates.detect().clusters
+    pixel = only(clusters, Method.PIXEL.value)
+    assert len(pixel) == 1
+    assert pixel[0]["confidence"] == "pixel"
+    assert pixel[0]["state"] == ClusterState.DECIDED.value
+
+
+def test_perceptual_variant_is_review_only(tmp_path, inventory, duplicates):
+    lib = tmp_path / "lib"
+    a = structured(lib / "a.jpg", 1)
+    resized_copy(a, lib / "a_small.jpg")
+    inventory.scan(lib)
+
+    clusters = duplicates.detect().clusters
+    perceptual = only(clusters, Method.PERCEPTUAL.value)
+    assert len(perceptual) == 1
+    assert perceptual[0]["state"] == ClusterState.OPEN.value  # never auto-decided
+    assert perceptual[0]["decision"] is None
+    assert perceptual[0]["confidence"] in ("near", "similar")
+
+
+def test_distinct_images_do_not_cluster(tmp_path, inventory, duplicates):
+    lib = tmp_path / "lib"
+    structured(lib / "a.jpg", 1)
+    structured(lib / "b.jpg", 2)
+    inventory.scan(lib)
+    assert duplicates.detect().clusters == []
+
+
+def test_not_duplicate_decision_persists_negative_link_and_suppresses(
+    tmp_path, inventory, duplicates, factory
+):
+    lib = tmp_path / "lib"
+    a = structured(lib / "a.jpg", 1)
+    resized_copy(a, lib / "a_small.jpg")
+    inventory.scan(lib)
+    cluster = only(duplicates.detect().clusters, Method.PERCEPTUAL.value)[0]
+
+    duplicates.decide(cluster["id"], Decision.NOT_DUPLICATE)
+
+    links = list(factory().execute(select(DuplicateNegativeLink)).scalars())
+    assert len(links) == 1
+
+    # Re-detecting must not re-suggest the rejected pair.
+    again = only(duplicates.detect().clusters, Method.PERCEPTUAL.value)
+    assert again == []
+
+
+def test_decision_is_reversible(tmp_path, inventory, duplicates, factory):
+    lib = tmp_path / "lib"
+    a = structured(lib / "a.jpg", 1)
+    resized_copy(a, lib / "a_small.jpg")
+    inventory.scan(lib)
+    cluster = only(duplicates.detect().clusters, Method.PERCEPTUAL.value)[0]
+    members = [m["asset_id"] for m in cluster["members"]]
+
+    duplicates.decide(cluster["id"], Decision.NOT_DUPLICATE)
+    assert factory().execute(select(DuplicateNegativeLink)).scalars().all()
+
+    # Reverse: choose a canonical instead.
+    reversed_snapshot = duplicates.decide(
+        cluster["id"], Decision.CANONICAL, canonical_asset_id=members[0]
+    )
+    assert reversed_snapshot["state"] == ClusterState.DECIDED.value
+    assert not factory().execute(select(DuplicateNegativeLink)).scalars().all()
+
+
+def test_version_conflict_is_rejected(tmp_path, inventory, duplicates):
+    lib = tmp_path / "lib"
+    a = structured(lib / "a.jpg", 1)
+    resized_copy(a, lib / "a_small.jpg")
+    inventory.scan(lib)
+    cluster = only(duplicates.detect().clusters, Method.PERCEPTUAL.value)[0]
+
+    with pytest.raises(ConflictError):
+        duplicates.decide(
+            cluster["id"], Decision.DEFERRED, expected_version=cluster["version"] + 5
+        )
+
+
+def test_canonical_must_be_a_member(tmp_path, inventory, duplicates):
+    lib = tmp_path / "lib"
+    a = structured(lib / "a.jpg", 1)
+    resized_copy(a, lib / "a_small.jpg")
+    inventory.scan(lib)
+    cluster = only(duplicates.detect().clusters, Method.PERCEPTUAL.value)[0]
+
+    with pytest.raises(DuplicateError):
+        duplicates.decide(cluster["id"], Decision.CANONICAL, canonical_asset_id="ghost")
+
+
+def test_no_canonical_cycles(tmp_path, inventory, duplicates, factory):
+    lib = tmp_path / "lib"
+    a = structured(lib / "a.jpg", 1)
+    shutil.copy2(a, lib / "a_copy.jpg")
+    inventory.scan(lib)
+    duplicates.detect()
+
+    # Every asset's canonical chain terminates (no loop).
+    assets = {a.id: a for a in factory().execute(select(Asset)).scalars()}
+    for start in assets:
+        seen, cur = set(), start
+        while cur is not None:
+            assert cur not in seen
+            seen.add(cur)
+            cur = assets[cur].canonical_asset_id
+
+
+def test_new_exact_member_inherits_decision(tmp_path, inventory, duplicates, factory):
+    lib = tmp_path / "lib"
+    a = structured(lib / "a.jpg", 1)
+    shutil.copy2(a, lib / "a2.jpg")
+    inventory.scan(lib)
+    cluster = only(duplicates.detect().clusters, Method.EXACT.value)[0]
+    canonical = cluster["canonical_asset_id"]
+
+    # A third exact copy appears later.
+    shutil.copy2(a, lib / "a3.jpg")
+    inventory.scan(lib)
+    cluster2 = only(duplicates.detect().clusters, Method.EXACT.value)[0]
+
+    assert cluster2["id"] == cluster["id"]
+    assert cluster2["state"] == ClusterState.DECIDED.value
+    assert len(cluster2["members"]) == 3
+    assets = {a.id: a for a in factory().execute(select(Asset)).scalars()}
+    variants = [m for m in cluster2["members"] if m["role"] == "variant"]
+    assert len(variants) == 2
+    for m in variants:
+        assert assets[m["asset_id"]].canonical_asset_id == canonical
+
+
+def test_contradictory_member_reopens_reviewed_cluster(
+    tmp_path, inventory, duplicates, db_url, make_factory
+):
+    lib = tmp_path / "lib"
+    a = structured(lib / "a.jpg", 1)
+    b = resized_copy(a, lib / "b.jpg")
+    inventory.scan(lib)
+    cluster = only(duplicates.detect().clusters, Method.PERCEPTUAL.value)[0]
+    duplicates.decide(cluster["id"], Decision.NOT_DUPLICATE)  # dismissed + negative link
+
+    # b is replaced by an exact copy of a — now byte-identical to a "not duplicate".
+    shutil.copy2(a, b)
+    inventory.scan(lib)
+    clusters = duplicates.detect().clusters
+
+    reopened = [c for c in clusters if c["state"] == ClusterState.REOPENED.value]
+    assert reopened, "byte-identical contradiction of a not_duplicate must reopen review"
+
+
+def test_decisions_durable_across_restart(tmp_path, db_url, make_factory):
+    lib = tmp_path / "lib"
+    a = structured(lib / "a.jpg", 1)
+    shutil.copy2(a, lib / "a_copy.jpg")
+
+    InventoryService(make_factory(db_url)).scan(lib)
+    cluster = only(
+        DuplicateService(make_factory(db_url)).detect().clusters, Method.EXACT.value
+    )[0]
+
+    # Fresh factory against the same database: the decision persisted.
+    from photo_pipeline.models import DuplicateCluster
+
+    with make_factory(db_url)() as session:
+        persisted = session.get(DuplicateCluster, cluster["id"])
+        assert persisted.state == ClusterState.DECIDED.value
+        assert persisted.decision == Decision.CANONICAL.value

tests/integration/test_migrations.py

@@ -7,6 +7,16 @@ import pytest
 
 from photo_pipeline.db import run_migrations
 
+
+def _alembic_head() -> str:
+    from alembic.config import Config as AlembicConfig
+    from alembic.script import ScriptDirectory
+
+    repo = Path(__file__).resolve().parents[2]
+    cfg = AlembicConfig(str(repo / "alembic.ini"))
+    cfg.set_main_option("script_location", str(repo / "migrations"))
+    return ScriptDirectory.from_config(cfg).get_current_head()
+
 # Faithful snapshot of the donor photo_analyzer.py schema (photos + FTS + triggers).
 LEGACY_SCHEMA = """
 CREATE TABLE photos (
@@ -50,7 +60,7 @@ def test_migrations_from_empty_database(tmp_path):
 
     tables = _tables(db)
     assert {"assets", "asset_paths"} <= tables
-    assert _head_revision(db) == "0001_initial_identity"
+    assert _head_revision(db) == _alembic_head()
 
 
 def test_migrations_from_legacy_snapshot_preserve_existing_data(tmp_path):
@@ -75,7 +85,7 @@ def test_migrations_from_legacy_snapshot_preserve_existing_data(tmp_path):
     finally:
         conn.close()
     assert row == ("album/one.jpg", "analyzed", "a red block")
-    assert _head_revision(db) == "0001_initial_identity"
+    assert _head_revision(db) == _alembic_head()
 
 
 def test_migrations_are_idempotent(tmp_path):
@@ -83,7 +93,7 @@ def test_migrations_are_idempotent(tmp_path):
     url = f"sqlite:///{db}"
     run_migrations(url)
     run_migrations(url)  # second run is a no-op at head
-    assert _head_revision(db) == "0001_initial_identity"
+    assert _head_revision(db) == _alembic_head()
 
 
 @pytest.mark.parametrize("expected", ["assets", "asset_paths"])

tests/unit/test_phash.py

@@ -0,0 +1,52 @@
+"""Golden perceptual-hash relationships and confidence bands."""
+
+import numpy as np
+from PIL import Image
+
+from photo_pipeline.services import hashing
+from photo_pipeline.services.duplicates import NEAR_MAX, SIMILAR_MAX
+
+
+def _structured(path, seed, size=(256, 192)):
+    rng = np.random.default_rng(seed)
+    w, h = size
+    base = np.zeros((h, w, 3), dtype=np.uint8)
+    for _ in range(6):
+        x0 = int(rng.integers(0, w - 60))
+        y0 = int(rng.integers(0, h - 60))
+        base[y0 : y0 + 60, x0 : x0 + 60] = rng.integers(0, 256, 3)
+    grad = np.linspace(0, 120, w, dtype=np.uint8)
+    base[:, :, 0] = np.clip(base[:, :, 0].astype(int) + grad[None, :], 0, 255)
+    Image.fromarray(base).save(path, quality=95)
+    return path
+
+
+def test_phash_is_16_hex_chars(tmp_path):
+    value = hashing.phash(_structured(tmp_path / "a.jpg", 1))
+    assert len(value) == 16
+    int(value, 16)  # parses as hex
+
+
+def test_identical_pixels_have_zero_distance(tmp_path):
+    a = _structured(tmp_path / "a.jpg", 1)
+    assert hashing.phash_distance(hashing.phash(a), hashing.phash(a)) == 0
+
+
+def test_resized_copy_stays_within_near_band(tmp_path):
+    a = _structured(tmp_path / "a.jpg", 1)
+    small = tmp_path / "a_small.jpg"
+    with Image.open(a) as image:
+        image.resize((image.width // 2, image.height // 2), Image.LANCZOS).save(small, quality=95)
+    distance = hashing.phash_distance(hashing.phash(a), hashing.phash(small))
+    assert distance <= NEAR_MAX
+
+
+def test_distinct_images_exceed_similar_band(tmp_path):
+    a = _structured(tmp_path / "a.jpg", 1)
+    b = _structured(tmp_path / "b.jpg", 2)
+    distance = hashing.phash_distance(hashing.phash(a), hashing.phash(b))
+    assert distance > SIMILAR_MAX
+
+
+def test_phash_is_versioned():
+    assert hashing.PHASH_VERSION >= 1