"""Phase 148, task 01 — render qualifying PDF pages at index time (unit
matrix).

The owner's TODO L5: "For primarily pictographic PDFs it's not super
helpful to see the summaries of the pages. I want to see the PDF pages
in the sources." The locked rule: a PDF gets rendered pages iff the
sync ALREADY decided its text is not the useful view — the phase-133
fallback fired (unusable text layer — the screenshot class) OR the
phase-140 pictorial pass described ≥1 page (the sparse-page class —
the IKEA fixture). Text-heavy PDFs (usable layer, zero candidates)
keep the text view, and ``BOR_PDF_PICTORIAL=0`` keeps its kill-switch
semantics (no pictorial descriptions → no page strip for the sparse
class).

Pinned here (the house PyMuPDF fixture recipes — the
``test_pictorial_import.py`` builder for the pictorial class, the
``test_importer.py`` garbage builder for the fallback class, the dense
builder for the negative; the copies land in a tmp ``pdf_page_dir``):

* eligibility × new index — the pictorial class (IKEA shape: usable
  layer, ≥1 sparse page) and the fallback class (unusable layer) get
  ``<doc-id>-p{N}.png`` for EVERY page (not just the described
  candidates — the strip shows the whole manual) + ``page_count`` =
  the page count + real PNG bytes; a text-heavy PDF gets neither
  (no copies, NULL count, ZERO render calls, byte-identical content);
* the phase-147 fast path — an unchanged qualifying PDF's second sync
  re-renders NOTHING (zero render calls — the copies persist,
  ``page_count`` and the bytes untouched);
* stale replacement — a changed (shrunk) PDF's old copy set is
  replaced: the new set covers the new pages, the old ``p{N}`` tail
  is GONE (no orphans), ``page_count`` follows the new count;
* qualifier loss — a qualifying PDF whose new version gained a dense
  text layer loses its strip: ``page_count`` NULL + copies deleted;
* unopenable — a render that yields no pages (``render_pdf_pages`` →
  ``[]``): ``page_count`` NULL, no copies, NO exception, the doc is
  still indexed exactly as pre-phase (the row lands, the summary
  runs);
* per-page gap — one unreadable page (the render's ``b""``
  placeholder): the other copies are written, ``page_count`` keeps
  the FULL count (the gap is a display-level 404 — task 02 pins the
  route's side);
* fail-soft — the render raising (warning is the signal, ``page_count``
  NULL, no copies, the doc indexed fully — chunks + summary) and the
  WRITER raising (the copies this call wrote are removed — no partial
  orphans, ``page_count`` NULL, the doc indexed fully);
* prune — a qualifying doc's file removed from the source: the row
  goes and its page copies go with it (no orphans in the tmp dir).

The route + payload (task 02) and the viewer strip (task 03) are
pinned in their own task files; the E2E surface is
``tests/e2e/test_pdf_pages_in_sources.py`` (task 04).
"""
from __future__ import annotations

import asyncio
import logging
import uuid
from collections.abc import Callable, Sequence
from pathlib import Path
from typing import Any

import pymupdf as fitz
import pytest
from sqlalchemy import select

from app.config import Settings
from app.models import Document
from app.rag import importer
from app.rag.importer import _delete_page_copies, import_sources
from app.rag.pdf import extract_pdf_text
from tests.unit.test_importer import _build_garbage_pdf, _PageVisionLLM
from tests.unit.test_pictorial_import import (
    TITLE,
    _build_dense_pdf,
    _build_ikea_like_pdf,
    _insert_dense,
    _PictorialLLM,
)

#: The PNG signature (``render_pdf_pages``' output is ``tobytes("png")``).
PNG_MAGIC = b"\x89PNG\r\n\x1a\n"


def _settings(tmp_path: Path, **kwargs: Any) -> Settings:
    """Settings admitting the ``pdf`` import token with the phase-140
    knobs at their CODE DEFAULTS (``pdf_pictorial`` ON — LOCKED A3) and
    the phase-148 ``pdf_page_dir`` pinned to *tmp_path* (the copies
    under test never land in the real home); ``images`` off (the
    embedded-image pass is not in play — the fixture class is
    vector-drawn)."""
    kwargs.setdefault("_env_file", None)
    kwargs.setdefault("import_extensions", "md,txt,pdf")
    kwargs.setdefault("images", False)
    kwargs.setdefault("image_dir", str(tmp_path / "images"))
    kwargs.setdefault("pdf_page_dir", str(tmp_path / "pages"))
    return Settings(**kwargs)  # pyright: ignore[reportCallIssue]


def _pages_dir(tmp_path: Path) -> Path:
    return tmp_path / "pages"


def _doc(db, source: str, rel: str) -> Document:
    doc = db.scalar(
        select(Document).where(Document.source == source, Document.path == rel)
    )
    assert doc is not None, f"no documents row for ({source!r}, {rel!r})"
    return doc


def _cleanup_source(db, source: str) -> None:
    for doc in db.scalars(select(Document).where(Document.source == source)).all():
        db.delete(doc)
    db.commit()


def _build_ikea_shrunk(path: Path) -> None:
    """A SHRUNK IKEA-class PDF (the stale-replacement target): 6 pages
    — the title page, two dense warnings, three sparse ``Step N``
    markers — still qualifying (candidates: 1 and 4–6), but four pages
    shorter than :func:`_build_ikea_like_pdf`'s 10 (the old ``p7..p10``
    tail must not leak)."""
    doc = fitz.open()
    doc.new_page().insert_text((72, 72), "BESTA")
    _insert_dense(doc.new_page())
    _insert_dense(doc.new_page())
    for n in range(4, 7):
        doc.new_page().insert_text((72, 72), f"Step {n}")
    doc.set_metadata({"title": TITLE})
    doc.save(path)
    doc.close()


def _render_spy(
    monkeypatch: pytest.MonkeyPatch,
    *,
    mutate_full: Callable[[list[bytes]], list[bytes]] | None = None,
    raise_full: bool = False,
) -> list[Any]:
    """Count + steer the ``render_pdf_pages`` calls the importer makes
    (the pictorial pass's candidate render AND the phase-148
    full-page render both resolve the module global at call time):
    returns the list of *pages* arguments seen (a candidate list for
    the pass, ``None`` for the full render). *mutate_full* transforms
    the full-render result (e.g. blanks one page's slot);
    *raise_full* makes the full render raise. The candidate render
    always runs the real helper."""
    real = importer.render_pdf_pages
    calls: list[Any] = []

    def spy(pdf_path: Path, dpi: int = 150, pages: Sequence[int] | None = None) -> list[bytes]:
        calls.append(pages)
        if pages is not None:
            return real(pdf_path, dpi=dpi, pages=pages)
        if raise_full:
            raise RuntimeError("simulated render failure (test sentinel)")
        out = real(pdf_path, dpi=dpi, pages=pages)
        return mutate_full(out) if mutate_full is not None else out

    monkeypatch.setattr(importer, "render_pdf_pages", spy)
    return calls


# ---------------------------------------------------------------------------
# Eligibility × new index
# ---------------------------------------------------------------------------


def test_pictorial_pdf_new_index_writes_every_page_copy(db, tmp_path: Path) -> None:
    """Eligible — the PICTORIAL class (IKEA shape: usable layer, ≥1
    sparse page): the first index writes EVERY page of the PDF (not
    just the described candidates — the strip shows the whole manual)
    as ``<doc-id>-p{N}.png`` under ``pdf_page_dir`` (N 1-based), sets
    ``page_count`` to the page count, and the files carry real PNG
    bytes — while the indexed content is unchanged by this phase (the
    phase-140 illustration blocks stay: the descriptions help the
    embeddings and stay in the DB for Q&A — they are simply not
    displayed, task 03)."""
    root = tmp_path / "pgp"
    root.mkdir()
    _build_ikea_like_pdf(root / "ikea.pdf")
    llm = _PictorialLLM()
    llm.settings = _settings(tmp_path)
    try:
        summary = asyncio.run(import_sources([root], llm, session=db))
        assert (summary.files, summary.added, summary.errors) == (1, 1, 0)
        assert summary.pdfs_pictorial == 1  # the pass ran (the eligibility class)
        doc = _doc(db, "pgp", "ikea.pdf")
        assert doc.page_count == 10, "page_count = the PDF's page count"
        page_dir = _pages_dir(tmp_path)
        for n in range(1, 11):
            path = page_dir / f"{doc.id}-p{n}.png"
            assert path.is_file(), f"copy p{n} is missing"
            assert path.read_bytes()[:8] == PNG_MAGIC, f"p{n} is not a real PNG"
        # No behavior change: the final content still carries the
        # phase-140 illustration blocks on the sparse pages (1, 6–10)
        # and not on the dense ones (2–5).
        assert "[Page 1 illustration:" in doc.content
        assert "[Page 6 illustration:" in doc.content
        assert "[Page 2 illustration:" not in doc.content
    finally:
        _cleanup_source(db, "pgp")


def test_fallback_class_garbage_pdf_writes_every_page_copy(db, tmp_path: Path) -> None:
    """Eligible — the FALLBACK class (unusable text layer — the
    screenshot class): the garbage PDF takes the phase-133 screenshot
    fallback UNCHANGED and now gets the page strip too — one copy per
    page + ``page_count`` (the viewer can show the actual pages of the
    screenshot class as well)."""
    root = tmp_path / "pgf"
    root.mkdir()
    _build_garbage_pdf(root / "garbage.pdf")  # 2 pages
    llm = _PageVisionLLM()
    llm.settings = _settings(tmp_path)
    try:
        summary = asyncio.run(import_sources([root], llm, session=db))
        assert (summary.files, summary.added, summary.errors) == (1, 1, 0)
        assert summary.pdfs_screenshotted == 1  # the fallback fired (the class)
        doc = _doc(db, "pgf", "garbage.pdf")
        assert doc.page_count == 2
        page_dir = _pages_dir(tmp_path)
        for n in (1, 2):
            path = page_dir / f"{doc.id}-p{n}.png"
            assert path.is_file(), f"copy p{n} is missing"
            assert path.read_bytes()[:8] == PNG_MAGIC
        # No behavior change: the content is still the fallback's
        # vision descriptions (the page-anchored separator shape).
        assert doc.content.startswith("---PAGE-1---")
    finally:
        _cleanup_source(db, "pgf")


def test_text_heavy_pdf_keeps_the_text_view(
    db, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    """Ineligible — a TEXT-HEAVY PDF (usable layer, zero candidates):
    the text view stays — no copies, NULL ``page_count``, ZERO render
    calls (the pictorial pass no-ops before any render), and the
    content is byte-identical to ``extract_pdf_text`` (the
    pre-phase-140/148 pin)."""
    root = tmp_path / "pgd"
    root.mkdir()
    pdf = root / "clean.pdf"
    _build_dense_pdf(pdf)
    expected = extract_pdf_text(pdf)
    llm = _PictorialLLM()
    llm.settings = _settings(tmp_path)
    calls = _render_spy(monkeypatch)
    try:
        summary = asyncio.run(import_sources([root], llm, session=db))
        assert (summary.files, summary.added, summary.errors) == (1, 1, 0)
        assert summary.pdfs_pictorial == 0  # zero candidates — the pass no-ops
        doc = _doc(db, "pgd", "clean.pdf")
        assert doc.page_count is None, "a text-heavy PDF keeps the text view"
        assert not _pages_dir(tmp_path).exists(), "no copies — the dir is never created"
        assert calls == [], "zero render calls (no fallback, zero candidates)"
        assert doc.content == expected, "byte-identical to the extraction"
    finally:
        _cleanup_source(db, "pgd")


# ---------------------------------------------------------------------------
# Lifecycle: unchanged / changed / qualifier loss
# ---------------------------------------------------------------------------


def test_unchanged_second_sync_renders_nothing(
    db, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    """The phase-147 fast path + phase 148: an UNCHANGED qualifying
    PDF's second sync costs ZERO render calls (the copies persist — no
    re-render, no orphans), and ``page_count`` and every copy's bytes
    are untouched. The fast path returns before the branch, so the
    render block is skipped by construction."""
    root = tmp_path / "pgu"
    root.mkdir()
    _build_ikea_like_pdf(root / "ikea.pdf")
    llm = _PictorialLLM()
    llm.settings = _settings(tmp_path)
    try:
        calls = _render_spy(monkeypatch)
        s1 = asyncio.run(import_sources([root], llm, session=db))
        assert (s1.added, s1.unchanged) == (1, 0)
        doc1 = _doc(db, "pgu", "ikea.pdf")
        assert doc1.page_count == 10
        assert len([c for c in calls if c is None]) == 1, (
            "exactly one full-page render on the first index"
        )
        page_dir = _pages_dir(tmp_path)
        copies = {
            n: (page_dir / f"{doc1.id}-p{n}.png").read_bytes() for n in range(1, 11)
        }
        assert all(b[:8] == PNG_MAGIC for b in copies.values())
        embeds1 = len(llm.calls)

        s2 = asyncio.run(import_sources([root], llm, session=db))
        assert (s2.added, s2.updated, s2.unchanged) == (0, 0, 1)
        assert len(calls) == 2, "second sync: ZERO render calls (the fast path)"
        assert len(llm.calls) == embeds1, "no re-embed (the fast path's zero-work)"

        doc2 = _doc(db, "pgu", "ikea.pdf")
        assert doc2.id == doc1.id
        assert doc2.page_count == 10, "page_count untouched"
        for n in range(1, 11):
            assert (page_dir / f"{doc2.id}-p{n}.png").read_bytes() == copies[n], (
                f"copy p{n} moved on an unchanged sync"
            )
    finally:
        _cleanup_source(db, "pgu")


def test_changed_pdf_replaces_the_copy_set_without_orphans(
    db, tmp_path: Path,
) -> None:
    """A CHANGED (shrunk, 10 → 6 pages) qualifying PDF: the re-index
    replaces the copy set — the new set covers p1..p6, the old
    version's ``p7..p10`` tail is GONE (the stale-replacement runs
    before the write), the doc keeps its id, and ``page_count``
    follows the new page count."""
    root = tmp_path / "pgc"
    root.mkdir()
    pdf = root / "ikea.pdf"
    _build_ikea_like_pdf(pdf)
    llm = _PictorialLLM()
    llm.settings = _settings(tmp_path)
    try:
        s1 = asyncio.run(import_sources([root], llm, session=db))
        assert s1.added == 1
        doc1 = _doc(db, "pgc", "ikea.pdf")
        assert doc1.page_count == 10

        _build_ikea_shrunk(pdf)  # same path, 6 pages — still qualifying
        s2 = asyncio.run(import_sources([root], llm, session=db))
        assert (s2.added, s2.updated, s2.unchanged) == (0, 1, 0)
        doc2 = _doc(db, "pgc", "ikea.pdf")
        assert doc2.id == doc1.id, "the doc keeps its id (the copy path is stable)"
        assert doc2.page_count == 6, "page_count follows the new page count"
        page_dir = _pages_dir(tmp_path)
        for n in range(1, 7):
            assert (page_dir / f"{doc2.id}-p{n}.png").is_file(), f"p{n} is missing"
        for n in range(7, 11):
            assert not (page_dir / f"{doc2.id}-p{n}.png").exists(), (
                f"stale p{n} from the 10-page version leaked"
            )
        leftovers = sorted(p.name for p in page_dir.iterdir())
        assert len(leftovers) == 6, "exactly the new copy set, no orphans"
    finally:
        _cleanup_source(db, "pgc")


def test_qualifier_loss_clears_page_count_and_copies(
    db, tmp_path: Path,
) -> None:
    """A qualifying PDF whose NEW VERSION gained a dense text layer
    (zero candidates, ``pdf_pictorial`` ON): the document STOPPED
    qualifying — ``page_count`` is NULL, the copies are deleted, and
    the text view returns (the content is the extraction,
    byte-identical)."""
    root = tmp_path / "pgl"
    root.mkdir()
    pdf = root / "ikea.pdf"
    _build_ikea_like_pdf(pdf)
    llm = _PictorialLLM()
    llm.settings = _settings(tmp_path)
    try:
        s1 = asyncio.run(import_sources([root], llm, session=db))
        assert s1.added == 1
        doc1 = _doc(db, "pgl", "ikea.pdf")
        assert doc1.page_count == 10
        assert all(
            _pages_dir(tmp_path).joinpath(f"{doc1.id}-p{n}.png").is_file()
            for n in range(1, 11)
        )

        _build_dense_pdf(pdf)  # same path: dense text layer, zero candidates
        s2 = asyncio.run(import_sources([root], llm, session=db))
        assert (s2.added, s2.updated, s2.unchanged) == (0, 1, 0)
        doc2 = _doc(db, "pgl", "ikea.pdf")
        assert doc2.id == doc1.id
        assert doc2.page_count is None, "a doc that stopped qualifying loses its strip"
        assert not list(_pages_dir(tmp_path).glob(f"{doc2.id}-p*.png")), (
            "the copies must be deleted on qualifier loss"
        )
        assert doc2.content == extract_pdf_text(pdf), "the text view returns"
    finally:
        _cleanup_source(db, "pgl")


# ---------------------------------------------------------------------------
# Fail-soft: unopenable / gap / render exception / write exception
# ---------------------------------------------------------------------------


def test_unopenable_pdf_page_render_is_fail_soft(
    db, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    """A render that yields NO pages (an unopenable file —
    ``render_pdf_pages`` → ``[]``): ``page_count`` NULL, no copies,
    NO exception — the document still indexes exactly as pre-phase
    (the fallback's all-failed path: an empty-content row lands, the
    summary runs)."""
    root = tmp_path / "pgn"
    root.mkdir()
    _build_garbage_pdf(root / "garbage.pdf")  # the fallback class
    llm = _PageVisionLLM()
    llm.settings = _settings(tmp_path)
    monkeypatch.setattr(importer, "render_pdf_pages", lambda *a, **k: [])
    try:
        summary = asyncio.run(import_sources([root], llm, session=db))
        assert (summary.files, summary.added, summary.errors) == (1, 1, 0)
        doc = _doc(db, "pgn", "garbage.pdf")
        assert doc.page_count is None, "no pages rendered — no strip"
        assert not _pages_dir(tmp_path).exists(), "no copies — the dir is never created"
        assert doc.content == "", "the fallback's all-failed render lands empty content"
        assert summary.summaries == 1, "the document is indexed exactly as pre-phase"
    finally:
        _cleanup_source(db, "pgn")


def test_per_page_gap_writes_the_other_copies_and_keeps_the_full_count(
    db, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    """One unreadable page (the render's ``b""`` placeholder — page 4
    here): a per-page failure is NOT an exception — the other copies
    are written, ``page_count`` keeps the FULL count, and the gap's
    file is simply absent (a display-level 404 — the route of task 02
    pins its side)."""
    root = tmp_path / "pgg"
    root.mkdir()
    _build_ikea_like_pdf(root / "ikea.pdf")
    llm = _PictorialLLM()
    llm.settings = _settings(tmp_path)

    def gap_page_four(pages: list[bytes]) -> list[bytes]:
        pages[3] = b""  # index 3 = page 4 (the helper's list-index contract)
        return pages

    _render_spy(monkeypatch, mutate_full=gap_page_four)
    try:
        asyncio.run(import_sources([root], llm, session=db))
        doc = _doc(db, "pgg", "ikea.pdf")
        assert doc.page_count == 10, "the gap does not shrink the count"
        page_dir = _pages_dir(tmp_path)
        for n in range(1, 11):
            path = page_dir / f"{doc.id}-p{n}.png"
            if n == 4:
                assert not path.exists(), "the gap has no file"
            else:
                assert path.is_file(), f"copy p{n} is missing"
    finally:
        _cleanup_source(db, "pgg")


def test_render_exception_is_fail_soft(
    db,
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
    caplog: pytest.LogCaptureFixture,
) -> None:
    """The full-page render RAISES: the warning is the signal,
    ``page_count`` NULL, no copies — and the document indexes EXACTLY
    as pre-phase (the pictorial content, the chunks, the summary) —
    the render is a display concern, it never gates the indexing."""
    root = tmp_path / "pge"
    root.mkdir()
    _build_ikea_like_pdf(root / "ikea.pdf")
    llm = _PictorialLLM()
    llm.settings = _settings(tmp_path)
    _render_spy(monkeypatch, raise_full=True)
    try:
        with caplog.at_level(logging.WARNING, logger="app.importer"):
            summary = asyncio.run(import_sources([root], llm, session=db))
        assert (summary.files, summary.added, summary.errors) == (1, 1, 0)
        doc = _doc(db, "pge", "ikea.pdf")
        assert doc.page_count is None
        assert not _pages_dir(tmp_path).exists(), "no copies — no partial orphans"
        assert doc.chunks, "the chunk pipeline ran (fail-soft)"
        assert summary.summaries == 1, "the summary ran (fail-soft)"
        assert "[Page 6 illustration:" in doc.content, (
            "the pictorial pass's content is untouched by the render failure"
        )
        assert any("PDF page render failed" in m for m in caplog.messages), (
            caplog.messages
        )
    finally:
        _cleanup_source(db, "pge")


def test_write_exception_is_fail_soft_and_leaves_no_orphans(
    db,
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
    caplog: pytest.LogCaptureFixture,
) -> None:
    """The PNG WRITER raises (a full disk) after the FIRST copy
    landed: the copies this call wrote are removed (no partial
    orphans), ``page_count`` NULL, the warning is the signal, and the
    document indexes exactly as pre-phase."""
    root = tmp_path / "pgw"
    root.mkdir()
    _build_ikea_like_pdf(root / "ikea.pdf")
    llm = _PictorialLLM()
    llm.settings = _settings(tmp_path)
    original_write = Path.write_bytes
    written_paths: list[Path] = []

    def boom(self: Path, data: bytes) -> int:
        written_paths.append(self)
        if len(written_paths) > 1:
            raise OSError("simulated disk full (test sentinel)")
        return original_write(self, data)

    monkeypatch.setattr(Path, "write_bytes", boom)  # pyright: ignore[reportArgumentType]
    try:
        with caplog.at_level(logging.WARNING, logger="app.importer"):
            summary = asyncio.run(import_sources([root], llm, session=db))
        assert (summary.files, summary.added, summary.errors) == (1, 1, 0)
        assert len(written_paths) == 2, "p1 landed, p2's write failed"
        doc = _doc(db, "pgw", "ikea.pdf")
        assert doc.page_count is None
        assert not list(_pages_dir(tmp_path).glob(f"{doc.id}-p*.png")), (
            "the partial write was cleaned up — no orphans"
        )
        assert not written_paths[0].exists(), "the landed p1 copy was removed"
        assert summary.summaries == 1, "the document is indexed exactly as pre-phase"
        assert any("PDF page render failed" in m for m in caplog.messages), (
            caplog.messages
        )
    finally:
        _cleanup_source(db, "pgw")


# ---------------------------------------------------------------------------
# The shared cleanup helper (the _delete_image_copy convention)
# ---------------------------------------------------------------------------


def test_delete_page_copies_helper_is_best_effort(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
    caplog: pytest.LogCaptureFixture,
) -> None:
    """The helper (the ``_delete_image_copy`` convention): the row
    knows its own extent — copies 1..``page_count`` are deleted, a
    NULL/zero count is a no-op, missing files are a SILENT no-op
    (already gone — e.g. the owner cleaned the pages dir), and an
    unreadable file is logged, never raised (copy cleanup must not
    break the sync)."""
    page_dir = tmp_path / "pages"
    page_dir.mkdir()
    doc_id = uuid.uuid4()
    doc = Document(
        id=doc_id,
        source="s",
        path="p.pdf",
        full_path="/p.pdf",
        title="t",
        content="c",
        content_hash="x",
        page_count=3,
    )
    p1 = page_dir / f"{doc_id}-p1.png"
    p3 = page_dir / f"{doc_id}-p3.png"
    p1.write_bytes(b"png1")
    p3.write_bytes(b"png3")  # p2 is missing (the silent no-op)

    original_unlink = Path.unlink

    def boom(self: Path, **kwargs: Any) -> None:
        if self == p3:
            raise OSError("simulated unreadable file (test sentinel)")
        return original_unlink(self, **kwargs)

    monkeypatch.setattr(Path, "unlink", boom)  # pyright: ignore[reportArgumentType]
    with caplog.at_level(logging.WARNING, logger="app.importer"):
        _delete_page_copies(doc, page_dir)
    assert not p1.exists(), "the existing copy was deleted"
    assert p3.exists(), "the unreadable file survived (cleanup never raises)"
    assert any(
        "could not delete PDF page copy" in m for m in caplog.messages
    ), caplog.messages

    # A NULL count is a no-op (fresh docs, non-qualifying docs, every
    # fail-soft miss) — no exception, no attempt.
    doc.page_count = None
    _delete_page_copies(doc, page_dir)
    assert p3.exists()


# ---------------------------------------------------------------------------
# Prune
# ---------------------------------------------------------------------------


def test_prune_deletes_page_copies(db, tmp_path: Path) -> None:
    """A qualifying doc is PRUNED (its file removed from the source):
    the row goes AND its page copies go with it — no orphans in the
    pages dir (the ``_prune`` hook beside the image-copy deletion)."""
    root = tmp_path / "pgz"
    root.mkdir()
    pdf = root / "ikea.pdf"
    _build_ikea_like_pdf(pdf)
    llm = _PictorialLLM()
    llm.settings = _settings(tmp_path)
    try:
        s1 = asyncio.run(import_sources([root], llm, session=db))
        assert s1.added == 1
        doc = _doc(db, "pgz", "ikea.pdf")
        assert doc.page_count == 10
        assert len(list(_pages_dir(tmp_path).glob(f"{doc.id}-p*.png"))) == 10

        pdf.unlink()
        s2 = asyncio.run(import_sources([root], llm, session=db, prune=True))
        assert s2.pruned == 1
        assert (
            db.scalar(
                select(Document).where(
                    Document.source == "pgz", Document.path == "ikea.pdf"
                )
            )
            is None
        ), "the row is pruned"
        assert not list(_pages_dir(tmp_path).glob("*.png")), (
            "no orphaned page copies after the prune"
        )
    finally:
        _cleanup_source(db, "pgz")