feat: detect books that may already be in the library

Match on a shared identifier, or on a normalised title credited to a shared
author, and report candidates rather than refusing anything — a metadata match
is a guess, and a second edition is not a mistake. Adds a library-wide review
pass, dismissals, and renames the fingerprint pre-flight to duplicate-files.

backend/AGENTS.md

@@ -149,8 +149,98 @@ Three things to preserve when touching this code:
   submitted twice in one request are caught. Those duplicates report `book_id: None` —
   there is no row to point at yet.
 
-`POST /books/duplicates` answers the same question from fingerprints alone, for clients
+`POST /books/duplicate-files` answers the same question from fingerprints alone, for
-that want to ask before uploading anything.
+clients that want to ask before uploading anything.
+
+### Book-level detection — a different question
+
+The file check answers "are these the same bytes?". `find_duplicate_books` answers "is
+this the same book?", which a re-scan, a re-zipped EPUB or another edition cannot be
+asked with a hash. Two signals, either sufficient: a **shared identifier**, or a
+**matching normalized title with at least one shared author**.
+
+It **never blocks**. A metadata match is a guess — a work shares title and author with
+its own translation, its own second edition and its own audiobook — so the book is
+created and the candidates are reported alongside it in
+`ImportResult.possible_duplicates`. File-level dedupe keeps its refuse/skip behaviour;
+that one is near-certain and this one is not. Do not "improve" this into a refusal.
+
+Two rules narrow it, both applied in Python over the small candidate set:
+
+- **A shared author is required for a title match.** Without it every book the
+  extractors gave up on and titled `Unknown` is a duplicate of every other one. A book
+  with no authors can therefore only match on an identifier.
+- **The same series at a different `series_position` disqualifies a match.** A trilogy
+  shares an author and often most of its title; the position is the library saying
+  outright that these are two books.
+
+A book is compared under **several title keys, not one** (`_title_keys`). Ebook files
+are overwhelmingly named `Title - Author.epub`, and wherever nothing inside the file
+overrode that name the author ended up in the title column — so one copy is stored as
+`Building Microservices` and another as `Building Microservices Sam Newman`. Both
+directions are generated, the author stripped off and the author added on, which is why
+the query is `normalized_title.in_(keys)` rather than `==`. This is still exact matching
+on an indexed column: no similarity score, nothing to tune. It does not weaken the
+shared-author requirement, which is a separate condition.
+
+`find_duplicate_book_groups` is the library-wide pass behind
+`GET /books/duplicate-books`, since the import-time check says nothing about a
+collection someone already has. It buckets books by every key they carry and merges the
+buckets with union-find, so A~B by ISBN and B~C by title land in one group. Pairs in
+`duplicate_dismissals` are never merged — a reader disagreeing with one pairing must not
+silently break a group that stands on other evidence.
+
+### Author names have one stored form
+
+`Author.name` is always the canonical form, produced by `format_author_name`. Extractors
+hand over whatever the file said — `Newman, Sam;` from a `DC:creator` list, `Sam Newman`
+from a PDF, `Sam Newman.epub` from a filename — and storing those verbatim is how one
+person becomes four rows in the sidebar, four entries in the author filter, and four
+books that never look like each other.
+
+This is **display** canonicalization, distinct from `normalize_author`, which throws
+away case, accents and spacing to build a comparison key nobody sees. Tidying only
+removes what an extractor added: a trailing separator, a file extension, and the
+`Surname, Given` ordering. It never touches case or accents — `Michał Płachta` and
+`Steve McConnell` are the author's own spelling, not something to correct.
+
+Three places have to agree, and `Author` keeps them together: the `@validates("name")`
+hook, `unique_hash`, and `unique_filter`. `as_unique_async` looks a row up with the
+filter and then constructs with the validator, so if the lookup used the raw name and
+the insert used the tidy one, every variant spelling would miss the existing row and
+then collide with it on the unique index.
+
+A form the rule does not recognise is **left exactly as it was found** — `Dave Thomas,
+Andy Hunt` is two people in one string, and flipping it would invent a third. Leaving a
+mess visible beats rewriting it wrongly.
+
+### The normalized columns are written by validators
+
+`Book.normalized_title`, `Author.normalized_name` and `Identifier.normalized_value` are
+derived from `services/matching.py` and kept current by **SQLAlchemy `@validates` hooks
+on the models**, not by any service. Assigning them directly is always wrong.
+
+This is deliberate and it is invisible at the call sites: `BookService` writes titles
+through at least three paths (`to_model_on_create`, `to_model_on_update`, and the
+`setattr` loop in `_populate_with_unique_relationships`), and a validator is the only
+thing a fourth cannot bypass. The columns carry plain, **non-unique** btree indexes —
+two spellings collapsing onto one value is the entire point.
+
+`services/matching.py` is imported *inside* those validators rather than at module
+scope: reaching it initialises the `chitai.services` package, which imports the
+services, which import the models. Keep the local import.
+
+The validators only fire on write, so a migration that adds one of these columns must
+backfill existing rows through the same helpers — see the `data_upgrades()` hook in
+`2026-08-15_add_book_matching_keys_and_duplicate__4358e7d4743a.py`.
+
+**Changing anything in `services/matching.py` needs a revision that recomputes them.**
+The keys are derived and already written, so a normalization change silently invalidates
+every stored row: a book written under the old rules just stops matching one written
+under the new rules, with nothing to show that anything is wrong. Copy
+`2026-08-15_recompute_book_matching_keys_ed41acf21270.py`, which exists because
+`normalize_title` learned to strip compact edition markers (`2E`, `5e`). It is
+idempotent and safe to re-run.
 
 ## KOReader hashing
 

backend/src/chitai/controllers/book.py

@@ -14,6 +14,7 @@ from litestar.response import File, Stream
 from litestar.exceptions import HTTPException
 from litestar.status_codes import (
     HTTP_200_OK,
+    HTTP_204_NO_CONTENT,
     HTTP_400_BAD_REQUEST,
     HTTP_409_CONFLICT,
 )
@@ -181,11 +182,86 @@ class BookController(Controller):
                 s.DuplicateFileRead.model_validate(duplicate)
                 for duplicate in result.duplicates
             ],
+            possible_duplicates=[
+                s.PossibleDuplicateRead.model_validate(possible)
+                for possible in result.possible_duplicates
+            ],
         )
 
+    @get(path="duplicate-books")
+    async def list_duplicate_books(
+        self, books_service: BookService, library: m.Library
+    ) -> list[s.DuplicateBookGroupRead]:
+        """
+        Report books already in the library that look like copies of one another.
+
+        The import-time check only ever sees what is arriving, so this is what covers
+        a collection someone already has. Matching is on metadata and therefore a
+        guess: a group is a question for the reader, not a verdict.
+
+        Query Parameters:
+            library_id: The library to review.
+
+        Injected Dependencies:
+            books_service: The book service for database operations.
+            library: The library to review.
+
+        Returns:
+            One entry per group of two or more books. Empty when there is nothing to
+            review, or when duplicate detection is switched off.
+        """
+        groups = await books_service.find_duplicate_book_groups(library)
+
+        return [
+            s.DuplicateBookGroupRead(
+                books=[s.DuplicateBookRead.model_validate(book) for book in group]
+            )
+            for group in groups
+        ]
+
+    @post(path="duplicate-books/dismissals", status_code=HTTP_204_NO_CONTENT)
+    async def dismiss_duplicate_books(
+        self, books_service: BookService, data: s.DuplicateDismissal
+    ) -> None:
+        """
+        Record that two books are not the same book.
+
+        Without this the review screen proposes the same wrong pair forever, which is
+        how a reader learns to stop looking at it.
+
+        Request Body:
+            data: The two book IDs. Order does not matter.
+
+        Injected Dependencies:
+            books_service: The book service for database operations.
+
+        Raises:
+            HTTPException: 400 if the two IDs are the same or either book is unknown.
+        """
+        try:
+            await books_service.dismiss_duplicates(data.book_a_id, data.book_b_id)
+        except ValueError as exc:
+            raise HTTPException(status_code=HTTP_400_BAD_REQUEST, detail=str(exc))
+
+    @delete(path="duplicate-books/dismissals")
+    async def restore_duplicate_books(
+        self, books_service: BookService, book_a_id: int, book_b_id: int
+    ) -> None:
+        """
+        Undo a dismissal, so the pair is proposed again.
+
+        Query Parameters:
+            book_a_id: One of the two books.
+            book_b_id: The other. Order does not matter.
+
+        Injected Dependencies:
+            books_service: The book service for database operations.
+        """
+        await books_service.restore_duplicates(book_a_id, book_b_id)
+
     # A question, not a change: 200 rather than the 201 a POST would default to.
-    @post(path="duplicates", status_code=HTTP_200_OK)
+    @post(path="duplicate-files", status_code=HTTP_200_OK)
-    async def check_duplicates(
+    async def check_duplicate_files(
         self,
         books_service: BookService,
         library: m.Library,

backend/src/chitai/schemas/__init__.py

@@ -6,8 +6,12 @@ from .book import (
     BooksCreateFromFiles,
     BooksUploadResult,
     BookMetadataUpdate,
+    DuplicateBookGroupRead,
+    DuplicateBookRead,
+    DuplicateDismissal,
     DuplicateFileRead,
     FileFingerprint,
+    PossibleDuplicateRead,
     FileMetadataRead,
     BookSeriesRead,
 )

backend/src/chitai/schemas/book.py

@@ -152,12 +152,59 @@ class DuplicateFileRead(BaseModel):
     book_title: str | None = None
 
 
+class DuplicateBookRead(BaseModel):
+    """
+    A stored book that may be the same book as another one.
+
+    Unlike `DuplicateFileRead` this is a guess: the evidence is metadata two editions
+    of one work legitimately share. Nothing was refused on the strength of it.
+    """
+
+    model_config = ConfigDict(from_attributes=True)
+
+    book_id: int
+    title: str
+    authors: list[str]
+    library_id: int
+    cover_image: Path | None = None
+
+    # Why it matched: "identifier" and/or "title-author".
+    matched_on: list[str]
+
+
+class PossibleDuplicateRead(BaseModel):
+    """A book that was imported, together with what it might be a second copy of."""
+
+    model_config = ConfigDict(from_attributes=True)
+
+    book_id: int
+    title: str
+    candidates: list[DuplicateBookRead]
+
+
+class DuplicateBookGroupRead(BaseModel):
+    """Books the library holds that all look like copies of one book."""
+
+    books: list[DuplicateBookRead]
+
+
+class DuplicateDismissal(BaseModel):
+    """Two books a reader is saying are not the same book."""
+
+    book_a_id: int
+    book_b_id: int
+
+
 class BooksUploadResult(BaseModel):
     """The outcome of a multi-file upload: what was created, and what was skipped."""
 
     created: list["BookRead"]
     skipped: list[DuplicateFileRead]
 
+    # Created, not skipped — these are books that went in and look like something the
+    # library already had. The reader decides what to do about it.
+    possible_duplicates: list[PossibleDuplicateRead] = Field(default_factory=list)
+
 
 class BookMetadataUpdate(BaseModel):
     title: str | None = None

backend/src/chitai/services/book.py

@@ -8,6 +8,7 @@ import mimetypes
 from collections.abc import Callable, Iterable, Sequence
 from dataclasses import dataclass, field, replace
 from io import BytesIO, RawIOBase
+from itertools import combinations
 from pathlib import Path
 import uuid
 import zipfile
@@ -23,7 +24,8 @@ from advanced_alchemy.service import (
 )
 from advanced_alchemy.repository import SQLAlchemyAsyncRepository
 from advanced_alchemy.filters import CollectionFilter
-from sqlalchemy import inspect, select, tuple_
+from sqlalchemy import and_, delete, inspect, or_, select, tuple_
+from sqlalchemy.orm import joinedload, selectinload
 from litestar.response import File
 from litestar.datastructures import UploadFile
 import aiofiles
@@ -37,6 +39,7 @@ from chitai.database.models import (
     Author,
     BookAuthorLink,
     BookTagLink,
+    DuplicateDismissal,
     Tag,
     Publisher,
     BookSeries,
@@ -46,6 +49,11 @@ from chitai.database.models import (
 )
 from chitai.schemas.book import BooksCreateFromFiles
 from chitai.services.filesystem_library import BookPathGenerator
+from chitai.services.matching import (
+    normalize_author,
+    normalize_identifier,
+    normalize_title,
+)
 from chitai.services.metadata_extractor import Extractor as MetadataExtractor
 from chitai.services.utils import (
     cleanup_empty_parent_directories,
@@ -92,6 +100,38 @@ class DuplicateFile:
     book_title: str | None = None
 
 
+# Why two books look like the same book. Both are reported when both apply.
+MATCHED_ON_IDENTIFIER = "identifier"
+MATCHED_ON_TITLE_AUTHOR = "title-author"
+
+
+@dataclass(frozen=True)
+class DuplicateBook:
+    """
+    A book already in the library that may be the same book as another one.
+
+    Unlike `DuplicateFile` this is a guess, not a fact: the evidence is metadata two
+    editions of one work legitimately share, and so do a work and its translation. It
+    is reported, never acted on.
+    """
+
+    book_id: int
+    title: str
+    authors: list[str]
+    library_id: int
+    matched_on: list[str]
+    cover_image: str | None = None
+
+
+@dataclass(frozen=True)
+class PossibleDuplicate:
+    """A book that was imported, together with what it might be a second copy of."""
+
+    book_id: int
+    title: str
+    candidates: list[DuplicateBook]
+
+
 @dataclass
 class ImportResult:
     """What an import produced: the books created, and the files left out of them."""
@@ -99,6 +139,10 @@ class ImportResult:
     books: list[Book] = field(default_factory=list)
     duplicates: list[DuplicateFile] = field(default_factory=list)
 
+    # Books that were created — nothing here was refused — and look like something the
+    # library already holds. The reader decides; see `find_duplicate_books`.
+    possible_duplicates: list[PossibleDuplicate] = field(default_factory=list)
+
 
 class DuplicateFilesError(Exception):
     """Raised when an import would store files that are already in the library."""
@@ -137,6 +181,180 @@ def _unused_path(path: Path) -> Path:
     return candidate
 
 
+def _title_keys(title: str, authors: set[str]) -> set[str]:
+    """
+    Every normalized title one book could reasonably be filed under.
+
+    Ebook files are overwhelmingly named `Title - Author.epub`, and wherever nothing
+    inside the file overrode that name, the author ended up in the title column. So
+    one copy of a book is stored as "Building Microservices" and another as
+    "Building Microservices Sam Newman", and comparing titles alone never puts the
+    two together — which is most of what this feature is for.
+
+    Both directions are generated, the author stripped off and the author added on,
+    so the comparison works whichever form each copy happens to hold. This stays an
+    exact match on an indexed column: no similarity score, nothing to tune.
+
+    Args:
+        title: The book's normalized title.
+        authors: Its normalized author keys.
+
+    Returns:
+        The title keys, or an empty set when there is no title to match on.
+    """
+    if not title:
+        return set()
+
+    keys = {title}
+
+    for author in authors:
+        if title.endswith(f" {author}"):
+            # A title that is only its author's name is not a title once stripped.
+            if stripped := title[: -len(author)].strip():
+                keys.add(stripped)
+        else:
+            keys.add(f"{title} {author}")
+
+    return keys
+
+
+def _matching_keys(data: Any) -> tuple[set[str], set[str], set[str]]:
+    """
+    The keys a book is compared on: its title, its authors, and its identifiers.
+
+    Accepts a `Book` — whose keys the model validators have already computed — or the
+    metadata dict an import is holding before any row exists.
+
+    Args:
+        data: A `Book`, or a dict with `title`, `authors` and `identifiers`.
+
+    Returns:
+        The set of title keys (empty if there is nothing to match on), the set of
+        author keys, and the set of identifier keys.
+    """
+    if isinstance(data, Book):
+        authors = {
+            author.normalized_name for author in data.authors if author.normalized_name
+        }
+        return (
+            _title_keys(data.normalized_title, authors),
+            authors,
+            {
+                identifier.normalized_value
+                for identifier in data.identifiers
+                if identifier.normalized_value
+            },
+        )
+
+    authors = set()
+    for author in data.get("authors") or []:
+        key = (
+            author.normalized_name
+            if isinstance(author, Author)
+            else normalize_author(author)
+        )
+        if key:
+            authors.add(key)
+
+    # Identifiers arrive as a `{name: value}` dict from the API and the extractors, and
+    # as `Identifier` rows once `_preprocess_book_data` has been over them.
+    raw = data.get("identifiers") or {}
+    pairs = (
+        raw.items()
+        if isinstance(raw, dict)
+        else ((identifier.name, identifier.value) for identifier in raw)
+    )
+
+    identifiers = {
+        key for name, value in pairs if (key := normalize_identifier(name, value))
+    }
+
+    title = normalize_title(data.get("title"))
+
+    return _title_keys(title, authors), authors, identifiers
+
+
+def _series_position(data: Any) -> tuple[str, str | None]:
+    """The series a book belongs to and where it sits in it, both normalized."""
+    if isinstance(data, Book):
+        series = data.series.title if data.series else None
+        position = data.series_position
+    else:
+        series = data.get("series")
+        series = series.title if isinstance(series, BookSeries) else series
+        position = data.get("series_position")
+
+    return normalize_title(series), (position or "").strip() or None
+
+
+def _is_different_volume(left: tuple[str, str | None], right: tuple[str, str | None]) -> bool:
+    """
+    Whether two books are numbered entries of one series, and not the same entry.
+
+    A trilogy shares an author, and often most of its title. Book two is emphatically
+    not a second copy of book one, and the position is the library saying so outright.
+
+    Args:
+        left: One book's `(series key, position)`, from `_series_position`.
+        right: The other book's.
+
+    Returns:
+        True when both name the same series at different positions.
+    """
+    (left_series, left_position), (right_series, right_position) = left, right
+
+    if not left_series or left_series != right_series:
+        return False
+
+    if left_position is None or right_position is None:
+        return False
+
+    try:
+        return float(left_position) != float(right_position)
+    except ValueError:
+        return left_position.casefold() != right_position.casefold()
+
+
+class _DisjointSet:
+    """
+    Union-find over book ids.
+
+    Books are bucketed by each key they carry, so one book reaches a group through its
+    ISBN and another through its title. Merging the buckets is what puts A, B and C in
+    one group when A matches B on an ISBN and B matches C on its title.
+    """
+
+    def __init__(self) -> None:
+        self._parent: dict[int, int] = {}
+
+    def find(self, item: int) -> int:
+        root = self._parent.setdefault(item, item)
+
+        while root != self._parent[root]:
+            root = self._parent[root]
+
+        while item != root:  # Flatten the path behind us.
+            self._parent[item], item = root, self._parent[item]
+
+        return root
+
+    def union(self, first: int, second: int) -> None:
+        left, right = self.find(first), self.find(second)
+        if left != right:
+            self._parent[right] = left
+
+    def groups(self) -> list[list[int]]:
+        """Every set of two or more members, in a stable order."""
+        grouped: dict[int, list[int]] = defaultdict(list)
+        for item in sorted(self._parent):
+            grouped[self.find(item)].append(item)
+
+        return sorted(
+            (members for members in grouped.values() if len(members) > 1),
+            key=lambda members: members[0],
+        )
+
+
 class _ZipStream(RawIOBase):
     """
     A write-only sink that hands whatever `ZipFile` writes back to the caller.
@@ -303,6 +521,266 @@ class BookService(SQLAlchemyAsyncRepositoryService[Book]):
 
         return matches
 
+    # Everything a candidate has to be able to answer for: what it matched on, and
+    # enough about itself to be shown to a reader.
+    _MATCH_LOADS = (
+        selectinload(Book.identifiers),
+        selectinload(Book.author_links).joinedload(BookAuthorLink.author),
+        joinedload(Book.series),
+    )
+
+    async def find_duplicate_books(
+        self,
+        data: ModelDictT[Book] | dict[str, Any],
+        library: Library,
+        exclude_book_id: int | None = None,
+    ) -> list[DuplicateBook]:
+        """
+        Look for books the library already holds that may be the same book as this one.
+
+        Two signals, either of which is enough: a shared identifier, or the same
+        normalized title credited to at least one of the same authors. Both are things
+        a second edition, a re-scan and a re-zipped EPUB carry when their bytes have
+        nothing in common, which is exactly the case file-level dedupe cannot see.
+
+        Nothing here refuses anything. The evidence is metadata a work shares with its
+        own translation and with its own second edition, so a wrong answer would cost a
+        reader a book they meant to keep.
+
+        Args:
+            data: The incoming book, as a `Book` or as the metadata dict an import is
+                holding before the row exists.
+            library: The library being imported into. The search is scoped to it unless
+                `duplicate_scope` widens or disables it.
+            exclude_book_id: A book to leave out — its own row, once it has one.
+
+        Returns:
+            One entry per candidate, in id order, saying what it matched on. Empty when
+            the incoming book carries nothing to match on: no identifier, and either no
+            title or no authors.
+        """
+        if settings.duplicate_scope == DuplicateScope.OFF:
+            return []
+
+        titles, authors, identifiers = _matching_keys(data)
+
+        conditions = []
+        if identifiers:
+            conditions.append(
+                Book.identifiers.any(Identifier.normalized_value.in_(identifiers))
+            )
+
+        # A shared author is required for a title match, never optional: without it
+        # every book the extractors gave up on and called "Unknown" is a duplicate of
+        # every other one.
+        if titles and authors:
+            conditions.append(
+                and_(
+                    Book.normalized_title.in_(titles),
+                    Book.authors.any(Author.normalized_name.in_(authors)),
+                )
+            )
+
+        if not conditions:
+            return []
+
+        statement = (
+            select(Book).where(or_(*conditions)).options(*self._MATCH_LOADS).order_by(Book.id)
+        )
+
+        if exclude_book_id is not None:
+            statement = statement.where(Book.id != exclude_book_id)
+
+        if settings.duplicate_scope == DuplicateScope.LIBRARY:
+            statement = statement.where(Book.library_id == library.id)
+
+        candidates = (
+            (await self.repository.session.execute(statement)).unique().scalars().all()
+        )
+
+        series = _series_position(data)
+        matches = []
+
+        for candidate in candidates:
+            if _is_different_volume(series, _series_position(candidate)):
+                continue
+
+            (
+                candidate_titles,
+                candidate_authors,
+                candidate_identifiers,
+            ) = _matching_keys(candidate)
+
+            matched_on = []
+            if identifiers & candidate_identifiers:
+                matched_on.append(MATCHED_ON_IDENTIFIER)
+            if titles & candidate_titles and authors & candidate_authors:
+                matched_on.append(MATCHED_ON_TITLE_AUTHOR)
+
+            if matched_on:
+                matches.append(self._describe_candidate(candidate, matched_on))
+
+        return matches
+
+    async def _load_for_matching(self, book_id: int) -> Book:
+        """
+        Fetch a book with everything the comparison reads already loaded.
+
+        A book handed back by `create` carries its relationships only as far as the
+        caller happened to populate them, and touching an unloaded one from async code
+        raises rather than lazy-loading. Asking for them outright is the whole fix.
+        """
+        statement = select(Book).where(Book.id == book_id).options(*self._MATCH_LOADS)
+
+        return (await self.repository.session.execute(statement)).unique().scalar_one()
+
+    async def find_duplicate_book_groups(
+        self, library: Library
+    ) -> list[list[DuplicateBook]]:
+        """
+        Group everything already stored that looks like more than one copy of one book.
+
+        `find_duplicate_books` only ever runs at import, so it says nothing about the
+        library someone already has. This is the pass over it, and the reason the
+        feature is worth having at all for an existing collection.
+
+        Books are bucketed by each key they carry and the buckets are then merged, so a
+        group holds everything transitively connected: A and B sharing an ISBN, B and C
+        sharing a title and an author, all three in one group. Pairs a reader has
+        dismissed are never merged, so disagreeing with one pairing does not silently
+        break a group that stands on other evidence.
+
+        Args:
+            library: The library to review. Scoped by `duplicate_scope`, exactly as the
+                import-time check is.
+
+        Returns:
+            Groups of two or more, in a stable order. Empty when detection is off.
+        """
+        if settings.duplicate_scope == DuplicateScope.OFF:
+            return []
+
+        statement = select(Book).options(*self._MATCH_LOADS).order_by(Book.id)
+
+        if settings.duplicate_scope == DuplicateScope.LIBRARY:
+            statement = statement.where(Book.library_id == library.id)
+
+        books = {
+            book.id: book
+            for book in (await self.repository.session.execute(statement))
+            .unique()
+            .scalars()
+            .all()
+        }
+
+        buckets: dict[tuple[str, str], list[int]] = defaultdict(list)
+        for book_id, book in books.items():
+            titles, authors, identifiers = _matching_keys(book)
+
+            for identifier in identifiers:
+                buckets[(MATCHED_ON_IDENTIFIER, identifier)].append(book_id)
+
+            # One bucket per (title key, author) pair. A book sits in several, so two
+            # copies meet as long as they agree on any one of them.
+            for title in titles:
+                for author in authors:
+                    buckets[(MATCHED_ON_TITLE_AUTHOR, f"{title}\x00{author}")].append(book_id)
+
+        dismissed = await self._dismissed_pairs()
+        series = {book_id: _series_position(book) for book_id, book in books.items()}
+
+        pairings = _DisjointSet()
+        matched_on: dict[int, set[str]] = defaultdict(set)
+
+        for (reason, _), members in buckets.items():
+            # Pairwise rather than wholesale: a dismissal and the series check both
+            # speak about two specific books, not about the bucket they landed in.
+            for left, right in combinations(members, 2):
+                if DuplicateDismissal.pair(left, right) in dismissed:
+                    continue
+                if _is_different_volume(series[left], series[right]):
+                    continue
+
+                pairings.union(left, right)
+                matched_on[left].add(reason)
+                matched_on[right].add(reason)
+
+        return [
+            [
+                self._describe_candidate(books[book_id], sorted(matched_on[book_id]))
+                for book_id in group
+            ]
+            for group in pairings.groups()
+        ]
+
+    async def dismiss_duplicates(self, book_a_id: int, book_b_id: int) -> None:
+        """
+        Record that two books are not the same book, and stop proposing them.
+
+        Args:
+            book_a_id: One of the books.
+            book_b_id: The other. Order does not matter; the pair is stored ordered.
+
+        Raises:
+            ValueError: If the two ids are the same, or either book does not exist.
+        """
+        pair = DuplicateDismissal.pair(book_a_id, book_b_id)
+
+        if pair[0] == pair[1]:
+            raise ValueError("A book cannot be dismissed against itself")
+
+        found = (
+            await self.repository.session.execute(select(Book.id).where(Book.id.in_(pair)))
+        ).scalars().all()
+
+        if len(set(found)) != 2:
+            raise ValueError("No such book")
+
+        if pair in await self._dismissed_pairs():
+            return
+
+        self.repository.session.add(
+            DuplicateDismissal(book_a_id=pair[0], book_b_id=pair[1])
+        )
+        await self.repository.session.flush()
+
+    async def restore_duplicates(self, book_a_id: int, book_b_id: int) -> None:
+        """
+        Undo a dismissal, so the pair is proposed again.
+
+        Args:
+            book_a_id: One of the books.
+            book_b_id: The other.
+        """
+        book_a_id, book_b_id = DuplicateDismissal.pair(book_a_id, book_b_id)
+
+        await self.repository.session.execute(
+            delete(DuplicateDismissal).where(
+                DuplicateDismissal.book_a_id == book_a_id,
+                DuplicateDismissal.book_b_id == book_b_id,
+            )
+        )
+
+    async def _dismissed_pairs(self) -> set[tuple[int, int]]:
+        """Every pair a reader has said is not a duplicate."""
+        rows = await self.repository.session.execute(
+            select(DuplicateDismissal.book_a_id, DuplicateDismissal.book_b_id)
+        )
+
+        return set(rows.all())
+
+    @staticmethod
+    def _describe_candidate(book: Book, matched_on: list[str]) -> DuplicateBook:
+        """Enough of a book to show a reader who has to decide whether it is the same."""
+        return DuplicateBook(
+            book_id=book.id,
+            title=book.title,
+            authors=[author.name for author in book.authors],
+            library_id=book.library_id,
+            matched_on=matched_on,
+            cover_image=book.cover_image,
+        )
+
     async def _reserve_book_path(
         self, parent: Path, book_id: int | None = None
     ) -> Path:
@@ -478,18 +956,45 @@ class BookService(SQLAlchemyAsyncRepositoryService[Book]):
             if not accepted:
                 continue
 
-            result.books.append(
+            book = await self.create_book(
-                await self.create_book(
                 {"files": accepted, "library_id": library.id},
                 library,
                 screen_duplicates=False,
                 fingerprints=fingerprints,
                 **kwargs,
             )
-            )
+            result.books.append(book)
+            await self._record_possible_duplicates(result, book, library)
 
         return result
 
+    async def _record_possible_duplicates(
+        self, result: ImportResult, book: Book, library: Library
+    ) -> None:
+        """
+        Note anything the library already holds that this book might be a copy of.
+
+        Run after the book is created, not instead of creating it: a metadata match is
+        a guess, and the cost of acting on a wrong one is refusing a legitimate second
+        edition or a translation. The row is flushed by now, so the book before it in
+        the same import is a candidate too.
+
+        Args:
+            result: The import being assembled, appended to in place.
+            book: The book that was just created.
+            library: The library it went into.
+        """
+        candidates = await self.find_duplicate_books(
+            await self._load_for_matching(book.id), library, exclude_book_id=book.id
+        )
+
+        if candidates:
+            result.possible_duplicates.append(
+                PossibleDuplicate(
+                    book_id=book.id, title=book.title, candidates=candidates
+                )
+            )
+
     async def create_many_from_existing_files(
         self,
         file_paths: list[Path],
@@ -597,7 +1102,10 @@ class BookService(SQLAlchemyAsyncRepositoryService[Book]):
 
             cleanup_empty_parent_directories(consume_path / group, consume_path)
 
-            result.books.append(await super().create(data))
+            book = await super().create(data)
+            result.books.append(book)
+            await self._record_possible_duplicates(result, book, library)
+
             await self.repository.session.commit()
 
         return result

backend/src/chitai/services/consume.py

@@ -126,6 +126,19 @@ class ConsumeDirectoryWatcher:
                     f"to {settings.duplicate_path}"
                 )
 
+            # Imported all the same — a metadata match is a guess, and there is nobody
+            # here to ask. The library's duplicates screen is where these get decided.
+            for possible in result.possible_duplicates:
+                names = ", ".join(
+                    f"{candidate.title} (#{candidate.book_id})"
+                    for candidate in possible.candidates
+                )
+                print(
+                    f"Imported {possible.title!r} (#{possible.book_id}), which may "
+                    f"already be in the library as: {names}"
+                )
+
+
         except Exception as e:
             print(f"Error processing batch: {e}")
             raise e

backend/tests/integration/test_book.py

@@ -570,7 +570,7 @@ class TestDuplicateHandling:
         stored = book["files"][0]
 
         response = await authenticated_client.post(
-            "/books/duplicates?library_id=1",
+            "/books/duplicate-files?library_id=1",
             json=[
                 {
                     "hash": stored["hash"],
@@ -590,6 +590,117 @@ class TestDuplicateHandling:
         assert matches[0]["book_id"] == book["id"]
 
 
+@pytest.mark.asyncio
+class TestDuplicateBooks:
+    """A second copy of a book is imported and reported, never refused."""
+
+    epub_path = Path("tests/data_files/Metamorphosis - Franz Kafka.epub")
+
+    def upload(self, name: str, pad: bool = False) -> list[tuple[str, tuple]]:
+        """
+        The fixture, optionally padded so it is a different file and the same book.
+
+        Padding the archive changes its size and its sampled hash without disturbing
+        the metadata, which is exactly the case the file-level check cannot see.
+        """
+        data = self.epub_path.read_bytes() + (b"\0" * 64 if pad else b"")
+
+        return [("files", (name, data, "application/epub+zip"))]
+
+    async def test_a_second_edition_is_created_and_reported(
+        self, authenticated_client: AsyncClient
+    ) -> None:
+        first = await authenticated_client.post(
+            "/books/fromFiles?library_id=1", files=self.upload("first.epub")
+        )
+        original = first.json()["created"][0]
+
+        second = await authenticated_client.post(
+            "/books/fromFiles?library_id=1", files=self.upload("second.epub", pad=True)
+        )
+
+        assert second.status_code == 201
+        body = second.json()
+
+        # Created, not skipped: a metadata match is a guess, and refusing a legitimate
+        # second edition costs more than a note does.
+        assert len(body["created"]) == 1
+        assert body["skipped"] == []
+
+        assert len(body["possible_duplicates"]) == 1
+        possible = body["possible_duplicates"][0]
+        assert possible["book_id"] == body["created"][0]["id"]
+
+        candidate = possible["candidates"][0]
+        assert candidate["book_id"] == original["id"]
+        assert candidate["title"] == original["title"]
+        assert candidate["authors"] == ["Franz Kafka"]
+        assert "title-author" in candidate["matched_on"]
+
+    async def test_the_review_screen_groups_them(
+        self, authenticated_client: AsyncClient
+    ) -> None:
+        first = await authenticated_client.post(
+            "/books/fromFiles?library_id=1", files=self.upload("first.epub")
+        )
+        second = await authenticated_client.post(
+            "/books/fromFiles?library_id=1", files=self.upload("second.epub", pad=True)
+        )
+
+        book_ids = sorted(
+            [first.json()["created"][0]["id"], second.json()["created"][0]["id"]]
+        )
+
+        response = await authenticated_client.get("/books/duplicate-books?library_id=1")
+
+        assert response.status_code == 200
+        groups = response.json()
+        assert len(groups) == 1
+        assert [book["book_id"] for book in groups[0]["books"]] == book_ids
+
+    async def test_a_dismissed_group_stays_dismissed(
+        self, authenticated_client: AsyncClient
+    ) -> None:
+        first = await authenticated_client.post(
+            "/books/fromFiles?library_id=1", files=self.upload("first.epub")
+        )
+        second = await authenticated_client.post(
+            "/books/fromFiles?library_id=1", files=self.upload("second.epub", pad=True)
+        )
+
+        pair = {
+            "book_a_id": first.json()["created"][0]["id"],
+            "book_b_id": second.json()["created"][0]["id"],
+        }
+
+        dismissed = await authenticated_client.post(
+            "/books/duplicate-books/dismissals", json=pair
+        )
+        assert dismissed.status_code == 204
+
+        response = await authenticated_client.get("/books/duplicate-books?library_id=1")
+        assert response.json() == []
+
+        restored = await authenticated_client.delete(
+            "/books/duplicate-books/dismissals"
+            f"?book_a_id={pair['book_b_id']}&book_b_id={pair['book_a_id']}"
+        )
+        assert restored.status_code == 204
+
+        response = await authenticated_client.get("/books/duplicate-books?library_id=1")
+        assert len(response.json()) == 1
+
+    async def test_dismissing_an_unknown_book_is_refused(
+        self, authenticated_client: AsyncClient
+    ) -> None:
+        response = await authenticated_client.post(
+            "/books/duplicate-books/dismissals",
+            json={"book_a_id": 1, "book_b_id": 9999},
+        )
+
+        assert response.status_code == 400
+
+
 # NOTE: the multi-book ZIP download is covered at the service level, in
 # tests/unit/test_services/test_book_service.py. Driving `/books/download` through
 # AsyncTestClient hangs in fixture teardown: it is the only `Stream` endpoint in the

backend/tests/unit/test_services/test_book_service.py

@@ -30,6 +30,21 @@ def upload(path: Path, name: str | None = None) -> UploadFile:
     )
 
 
+def edition(path: Path, name: str) -> UploadFile:
+    """
+    Another edition of one of the fixtures: same book, different bytes.
+
+    Padding the archive changes the size and the sampled hash without disturbing
+    anything a reader or a metadata extractor sees, which is the case file-level
+    dedupe cannot answer and book-level detection exists for.
+    """
+    return UploadFile(
+        content_type="application/epub+zip",
+        filename=name,
+        file_data=path.read_bytes() + b"\0" * 64,
+    )
+
+
 @pytest.mark.asyncio
 class TestBookServiceCRUD:
     """Test CRUD operation for libraries."""
@@ -640,3 +655,343 @@ class TestMissingFiles:
         assert len(book.files) == 1
         assert path.is_file()
         assert path.read_bytes() == EPUB.read_bytes()
+
+
+async def store_book(
+    books_service: BookService, library: m.Library, **metadata
+) -> m.Book:
+    """
+    A book in the database carrying exactly the metadata given, and no files.
+
+    Book-level matching reads metadata only, so going through the file pipeline would
+    just make every case depend on what a fixture EPUB happens to declare.
+    """
+    book = await books_service.to_model_on_create(
+        BookCreate(library_id=library.id, **metadata).model_dump()
+    )
+
+    books_service.repository.session.add(book)
+    await books_service.repository.session.commit()
+
+    return book
+
+
+@pytest.mark.asyncio
+class TestDuplicateBooks:
+    """The same book arriving as different bytes: reported, never refused."""
+
+    async def test_an_identifier_alone_is_enough(
+        self, books_service: BookService, test_library: m.Library
+    ) -> None:
+        """Two printings of one edition agree on the ISBN and nothing else."""
+        stored = await store_book(
+            books_service,
+            test_library,
+            title="Metamorphosis",
+            authors=["Franz Kafka"],
+            identifiers={"isbn-13": "9780486282114"},
+        )
+
+        matches = await books_service.find_duplicate_books(
+            {
+                "title": "Die Verwandlung",
+                "authors": ["Someone Else"],
+                # The ISBN-10 of the same edition, written with its hyphens.
+                "identifiers": {"isbn-10": "0-486-28211-2"},
+            },
+            test_library,
+        )
+
+        assert [match.book_id for match in matches] == [stored.id]
+        assert matches[0].matched_on == ["identifier"]
+        assert matches[0].authors == ["Franz Kafka"]
+
+    async def test_a_title_and_a_shared_author_are_enough(
+        self, books_service: BookService, test_library: m.Library
+    ) -> None:
+        """A re-scan carries no identifier at all, only what is on the cover."""
+        stored = await store_book(
+            books_service,
+            test_library,
+            title="The Metamorphosis",
+            authors=["Kafka, Franz"],
+        )
+
+        matches = await books_service.find_duplicate_books(
+            {"title": "Metamorphosis", "authors": ["Franz Kafka"]}, test_library
+        )
+
+        assert [match.book_id for match in matches] == [stored.id]
+        assert matches[0].matched_on == ["title-author"]
+
+    async def test_an_author_left_in_the_title_still_matches(
+        self, books_service: BookService, test_library: m.Library
+    ) -> None:
+        """
+        Files are named `Title - Author.epub`, and that name often became the title.
+
+        One copy stored as "Building Microservices" and another as "Building
+        Microservices Sam Newman" are the same book, and comparing the title columns
+        as they stand would never say so.
+        """
+        stored = await store_book(
+            books_service,
+            test_library,
+            title="Building Microservices - Sam Newman",
+            authors=["Sam Newman"],
+        )
+
+        matches = await books_service.find_duplicate_books(
+            {"title": "Building Microservices", "authors": ["Newman, Sam;"]}, test_library
+        )
+
+        assert [match.book_id for match in matches] == [stored.id]
+        assert matches[0].matched_on == ["title-author"]
+
+    async def test_the_author_in_the_title_works_the_other_way_round(
+        self, books_service: BookService, test_library: m.Library
+    ) -> None:
+        """Whichever copy arrived first, the comparison has to reach the other."""
+        stored = await store_book(
+            books_service,
+            test_library,
+            title="Building Microservices",
+            authors=["Newman, Sam;"],
+        )
+
+        matches = await books_service.find_duplicate_books(
+            {"title": "Building Microservices - Sam Newman", "authors": ["Sam Newman"]},
+            test_library,
+        )
+
+        assert [match.book_id for match in matches] == [stored.id]
+
+    async def test_a_shared_author_is_still_required(
+        self, books_service: BookService, test_library: m.Library
+    ) -> None:
+        """The title variants must not become a way around the author requirement."""
+        await store_book(
+            books_service,
+            test_library,
+            title="Building Microservices - Sam Newman",
+            authors=["Sam Newman"],
+        )
+
+        assert (
+            await books_service.find_duplicate_books(
+                {"title": "Building Microservices", "authors": ["Martin Fowler"]},
+                test_library,
+            )
+            == []
+        )
+
+    async def test_a_title_without_a_shared_author_is_not_enough(
+        self, books_service: BookService, test_library: m.Library
+    ) -> None:
+        """Otherwise every book the extractors gave up on matches every other one."""
+        await store_book(
+            books_service, test_library, title="Metamorphosis", authors=["Franz Kafka"]
+        )
+
+        assert (
+            await books_service.find_duplicate_books(
+                {"title": "Metamorphosis", "authors": ["Peter Kuper"]}, test_library
+            )
+            == []
+        )
+
+    async def test_a_book_with_no_authors_can_only_match_on_an_identifier(
+        self, books_service: BookService, test_library: m.Library
+    ) -> None:
+        await store_book(
+            books_service, test_library, title="Unknown", authors=["Franz Kafka"]
+        )
+
+        assert (
+            await books_service.find_duplicate_books(
+                {"title": "Unknown", "authors": []}, test_library
+            )
+            == []
+        )
+
+    async def test_another_volume_of_a_series_is_not_a_duplicate(
+        self, books_service: BookService, test_library: m.Library
+    ) -> None:
+        """The position is the library saying outright that these are two books."""
+        await store_book(
+            books_service,
+            test_library,
+            title="Foundation",
+            authors=["Isaac Asimov"],
+            series="Foundation",
+            series_position="1",
+        )
+
+        incoming = {
+            "title": "Foundation",
+            "authors": ["Isaac Asimov"],
+            "series": "Foundation",
+        }
+
+        assert await books_service.find_duplicate_books(
+            incoming | {"series_position": "2"}, test_library
+        ) == []
+
+        # The same volume, written a little differently, still matches.
+        assert len(
+            await books_service.find_duplicate_books(
+                incoming | {"series_position": "1.0"}, test_library
+            )
+        ) == 1
+
+    async def test_a_book_is_not_its_own_duplicate(
+        self, books_service: BookService, test_library: m.Library
+    ) -> None:
+        stored = await store_book(
+            books_service, test_library, title="Metamorphosis", authors=["Franz Kafka"]
+        )
+
+        assert (
+            await books_service.find_duplicate_books(
+                stored, test_library, exclude_book_id=stored.id
+            )
+            == []
+        )
+
+    async def test_scope_off_reports_nothing(
+        self,
+        books_service: BookService,
+        test_library: m.Library,
+        monkeypatch: pytest.MonkeyPatch,
+    ) -> None:
+        await store_book(
+            books_service, test_library, title="Metamorphosis", authors=["Franz Kafka"]
+        )
+        monkeypatch.setattr(settings, "duplicate_scope", DuplicateScope.OFF)
+
+        incoming = {"title": "Metamorphosis", "authors": ["Franz Kafka"]}
+
+        assert await books_service.find_duplicate_books(incoming, test_library) == []
+        assert await books_service.find_duplicate_book_groups(test_library) == []
+
+    async def test_an_import_reports_the_copy_it_just_made(
+        self, books_service: BookService, test_library: m.Library
+    ) -> None:
+        """Different bytes, same book — created, and said so."""
+        await books_service.create_many_from_files(
+            BooksCreateFromFiles(files=[upload(EPUB)]), test_library
+        )
+
+        result = await books_service.create_many_from_files(
+            BooksCreateFromFiles(files=[edition(EPUB, "second.epub")]), test_library
+        )
+
+        assert len(result.books) == 1
+        assert result.duplicates == []
+        assert len(result.possible_duplicates) == 1
+
+        possible = result.possible_duplicates[0]
+        assert possible.book_id == result.books[0].id
+        assert "title-author" in possible.candidates[0].matched_on
+
+
+@pytest.mark.asyncio
+class TestDuplicateBookGroups:
+    """The pass over a library someone already has."""
+
+    async def test_groups_join_transitively(
+        self, books_service: BookService, test_library: m.Library
+    ) -> None:
+        """A and B by ISBN, B and C by title and author, all three in one group."""
+        first = await store_book(
+            books_service,
+            test_library,
+            title="Metamorphosis",
+            authors=["Franz Kafka"],
+            identifiers={"isbn-13": "9780486282114"},
+        )
+        second = await store_book(
+            books_service,
+            test_library,
+            title="The Trial",
+            authors=["Franz Kafka"],
+            identifiers={"isbn-10": "0486282112"},
+        )
+        third = await store_book(
+            books_service, test_library, title="Trial", authors=["Kafka, Franz"]
+        )
+
+        groups = await books_service.find_duplicate_book_groups(test_library)
+
+        assert len(groups) == 1
+        assert [book.book_id for book in groups[0]] == [first.id, second.id, third.id]
+
+        reasons = {book.book_id: book.matched_on for book in groups[0]}
+        assert reasons[first.id] == ["identifier"]
+        assert reasons[second.id] == ["identifier", "title-author"]
+        assert reasons[third.id] == ["title-author"]
+
+    async def test_a_lone_book_is_not_a_group(
+        self, books_service: BookService, test_library: m.Library
+    ) -> None:
+        await store_book(
+            books_service, test_library, title="Metamorphosis", authors=["Franz Kafka"]
+        )
+
+        assert await books_service.find_duplicate_book_groups(test_library) == []
+
+    async def test_a_dismissed_pair_is_left_out(
+        self, books_service: BookService, test_library: m.Library
+    ) -> None:
+        """Disagreeing with one pairing must not break a group standing on another."""
+        first = await store_book(
+            books_service,
+            test_library,
+            title="Metamorphosis",
+            authors=["Franz Kafka"],
+            identifiers={"isbn-13": "9780486282114"},
+        )
+        second = await store_book(
+            books_service,
+            test_library,
+            title="The Trial",
+            authors=["Franz Kafka"],
+            identifiers={"isbn-10": "0486282112"},
+        )
+        third = await store_book(
+            books_service, test_library, title="Trial", authors=["Kafka, Franz"]
+        )
+
+        await books_service.dismiss_duplicates(second.id, first.id)
+
+        groups = await books_service.find_duplicate_book_groups(test_library)
+        assert [[book.book_id for book in group] for group in groups] == [
+            [second.id, third.id]
+        ]
+
+        await books_service.restore_duplicates(first.id, second.id)
+
+        groups = await books_service.find_duplicate_book_groups(test_library)
+        assert [[book.book_id for book in group] for group in groups] == [
+            [first.id, second.id, third.id]
+        ]
+
+    async def test_a_book_cannot_be_dismissed_against_itself(
+        self, books_service: BookService, test_library: m.Library
+    ) -> None:
+        stored = await store_book(
+            books_service, test_library, title="Metamorphosis", authors=["Franz Kafka"]
+        )
+
+        with pytest.raises(ValueError):
+            await books_service.dismiss_duplicates(stored.id, stored.id)
+
+    async def test_dismissing_an_unknown_book_is_refused(
+        self, books_service: BookService, test_library: m.Library
+    ) -> None:
+        stored = await store_book(
+            books_service, test_library, title="Metamorphosis", authors=["Franz Kafka"]
+        )
+
+        with pytest.raises(ValueError):
+            await books_service.dismiss_duplicates(stored.id, stored.id + 10_000)