brain-of-reese/tests/unit/test_pdf_page_copies.py

"""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 + the phase-163 CLASS stamp
  (``page_strip_mode`` ``additive`` / ``primary``); a text-heavy PDF
  gets neither (no copies, NULL count, NULL mode, 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);
* WRITE-FIRST replacement (phase 163, task 02, LOCKED A4) — a changed
  (shrunk) PDF's new copies land atomically (tmp + ``os.replace``,
  pinned) BEFORE any old copy is removed; the old ``p{N}`` tail is
  GONE only after the replacement bytes are durable (no orphans),
  ``page_count`` follows the new count; a MID-WRITE failure leaves the
  prior copy set intact (the rollback-orphan state — the row over
  zero files — is structurally unrepresentable);
* 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``
  + ``page_strip_mode`` NULL, no copies, the doc indexed fully — chunks
  + summary) and the WRITER raising (this call's ``.tmp`` files and the
  finals it CREATED are removed — no partial orphans, ``page_count`` +
  ``page_strip_mode`` 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 os
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


# ---------------------------------------------------------------------------
# Phase 163 (task 01) — the strip-class column beside page_count
# ---------------------------------------------------------------------------


def test_pictorial_index_stamps_the_additive_class(db, tmp_path: Path) -> None:
    """Phase 163 (task 02, re-cut of the task-01 pin) — the STAMP: a
    qualifying PDF indexed through the PICTORIAL class gets
    ``page_strip_mode == "additive"`` beside ``page_count`` (the text +
    toggle view); the strip's copies themselves are unchanged by the
    phase (task 01's shape, mode stamped exactly with the count).
    """
    root = tmp_path / "pgm"
    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)
        doc = _doc(db, "pgm", "ikea.pdf")
        assert doc.page_count == 10, "the strip itself is unchanged by task 02"
        assert doc.page_strip_mode == "additive", (
            "the pictorial class stamps the additive mode (LOCKED A1/A3)"
        )
    finally:
        _cleanup_source(db, "pgm")


# ---------------------------------------------------------------------------
# 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
        assert doc.page_strip_mode == "primary", (
            "the fallback class stamps the primary mode — the strip IS the "
            "view, byte-identical to phase 148 otherwise (LOCKED A1)"
        )
        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 doc.page_strip_mode is None, "no strip — no class either"
        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-tail delete runs
    LAST, after the write-first loop — phase 163, task 02), the doc
    keeps its id, ``page_count`` follows the new page count, and
    ``page_strip_mode`` is (re-)stamped."""
    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"
        assert doc2.page_strip_mode == "additive", "the class is (re-)stamped"
        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 doc2.page_strip_mode is None, "the class is cleared with the count"
        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 doc.page_strip_mode is None, "the class is cleared with the count"
        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 atomic copy
    landed: this call's ``.tmp`` files and the final-name files it
    CREATED are removed (no partial orphans — phase 163, task 02),
    ``page_count`` AND ``page_strip_mode`` 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)
        n = original_write(self, data)  # the tmp file really lands first
        if len(written_paths) > 1:
            raise OSError("simulated disk full (test sentinel)")
        return n

    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's tmp landed + replaced, p2's write failed"
        assert all(p.name.endswith(".png.tmp") for p in written_paths), (
            "the writer only ever touches .tmp paths (atomic replace)"
        )
        doc = _doc(db, "pgw", "ikea.pdf")
        assert doc.page_count is None
        assert doc.page_strip_mode is None, "the class is cleared with the count"
        assert not list(_pages_dir(tmp_path).glob(f"{doc.id}-p*.png")), (
            "the partial write was cleaned up — no orphans"
        )
        assert not list(_pages_dir(tmp_path).glob("*.tmp")), (
            "the failed call's .tmp file was cleaned up too"
        )
        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")


# ---------------------------------------------------------------------------
# Phase 163 (task 02) — WRITE-FIRST, atomic per file (LOCKED A4)
# ---------------------------------------------------------------------------


def test_write_first_new_bytes_are_durable_before_any_stale_delete(
    db, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    """THE order pin (LOCKED A4 — the 2026-10-07 live loss): re-rendering
    a CHANGED (shrunk, 10 → 6) PDF writes every replacement page FIRST
    (tmp + ``os.replace``) and only then deletes the stale tail — the
    event log shows ALL replaces of the fresh bytes before the FIRST
    unlink of an old-extent file, the overwritten pages keep the NEW
    bytes, the shrunk tail (p7..p10) is gone, and nothing else is
    deleted."""
    root = tmp_path / "pgord"
    root.mkdir()
    pdf = root / "ikea.pdf"
    _build_ikea_like_pdf(pdf)
    llm = _PictorialLLM()
    llm.settings = _settings(tmp_path)
    asyncio.run(import_sources([root], llm, session=db))
    doc1 = _doc(db, "pgord", "ikea.pdf")
    page_dir = _pages_dir(tmp_path)
    sentinel = b"OLD-COPY-SENTINEL"
    for n in range(1, 11):  # make the old copies unmistakably distinguishable
        (page_dir / f"{doc1.id}-p{n}.png").write_bytes(sentinel)

    events: list[tuple[str, str]] = []
    real_replace = os.replace

    def replace_spy(src, dst, *args, **kwargs):  # pyright: ignore[reportUnknownParameterType]
        events.append(("replace", Path(os.fspath(dst)).name))
        return real_replace(src, dst, *args, **kwargs)

    real_unlink = Path.unlink

    def unlink_spy(self: Path, **kwargs: Any) -> None:
        if self.parent == page_dir and self.name.startswith(str(doc1.id)):
            events.append(("unlink", self.name))
        return real_unlink(self, **kwargs)

    monkeypatch.setattr(importer.os, "replace", replace_spy)
    monkeypatch.setattr(Path, "unlink", unlink_spy)  # pyright: ignore[reportArgumentType]
    try:
        _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)
    finally:
        monkeypatch.undo()  # pyright: ignore[reportUnknownMemberType]
        _cleanup_source(db, "pgord")
    replaces = [name for kind, name in events if kind == "replace"]
    unlinks = [name for kind, name in events if kind == "unlink"]
    assert len(replaces) == 6, "the 6 fresh pages are written via os.replace"
    assert set(unlinks) == {f"{doc1.id}-p{n}.png" for n in range(7, 11)}, (
        "the stale-tail delete touches ONLY the old extent the render did "
        "not rewrite — a file about to be overwritten is never deleted"
    )
    first_unlink = next(i for i, e in enumerate(events) if e[0] == "unlink")
    last_replace = max(i for i, e in enumerate(events) if e[0] == "replace")
    assert last_replace < first_unlink, (
        f"write-first violated — event order: {events}"
    )
    for n in range(1, 7):
        blob = (page_dir / f"{doc1.id}-p{n}.png").read_bytes()
        assert blob != sentinel and blob[:8] == PNG_MAGIC, (
            f"p{n} keeps the NEW rendered bytes"
        )
    for n in range(7, 11):
        assert not (page_dir / f"{doc1.id}-p{n}.png").exists(), f"stale p{n} leaked"
    assert not list(page_dir.glob("*.tmp")), "no .tmp orphans"


def test_mid_write_failure_keeps_the_prior_copies_intact(
    db,
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
    caplog: pytest.LogCaptureFixture,
) -> None:
    """THE failure pin (LOCKED A4): the writer raises on the SECOND
    page's tmp write of a changed re-render — the prior copy set stays
    on disk INTACT (the rollback-orphan state — the row over zero
    files — is structurally unrepresentable): every one of the 10 prior
    files still exists (p1 with the new bytes already atomically
    replaced before the failure, p2..p10 byte-for-byte the prior
    copies), zero ``.tmp`` orphans from this call, and the caller's
    fail-soft fires (``page_count``/``page_strip_mode`` NULL + the
    warning, the doc indexed fully)."""
    root = tmp_path / "pgfail"
    root.mkdir()
    pdf = root / "ikea.pdf"
    _build_ikea_like_pdf(pdf)
    llm = _PictorialLLM()
    llm.settings = _settings(tmp_path)
    asyncio.run(import_sources([root], llm, session=db))
    doc1 = _doc(db, "pgfail", "ikea.pdf")
    page_dir = _pages_dir(tmp_path)
    prior = {
        n: (page_dir / f"{doc1.id}-p{n}.png").read_bytes() for n in range(1, 11)
    }
    # Tag the re-render's bytes so the new p1 is recognizable, then
    # blow up the SECOND tmp write (page 2).
    _render_spy(
        monkeypatch,
        mutate_full=lambda pages: [
            PNG_MAGIC + f"NEW-{i}".encode() for i in range(len(pages))
        ],
    )
    original_write = Path.write_bytes
    tmp_writes: list[Path] = []

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

    monkeypatch.setattr(Path, "write_bytes", boom)  # pyright: ignore[reportArgumentType]
    _build_ikea_shrunk(pdf)
    try:
        with caplog.at_level(logging.WARNING, logger="app.importer"):
            s2 = asyncio.run(import_sources([root], llm, session=db))
    finally:
        monkeypatch.undo()  # pyright: ignore[reportUnknownMemberType]
    assert (s2.added, s2.updated, s2.unchanged) == (0, 1, 0)
    assert len(tmp_writes) == 2, "p1's tmp wrote, p2's tmp write raised"
    doc2 = _doc(db, "pgfail", "ikea.pdf")
    assert doc2.page_count is None, "the caller's fail-soft nulled the count"
    assert doc2.page_strip_mode is None, "the caller's fail-soft nulled the class"
    assert any("PDF page render failed" in m for m in caplog.messages), caplog.messages
    # The prior set is INTACT: exactly the 10 names, none lost, none
    # orphaned on top.
    assert {p.name for p in page_dir.iterdir()} == {
        f"{doc1.id}-p{n}.png" for n in range(1, 11)
    }, "files on disk exactly the prior set"
    assert (
        page_dir / f"{doc1.id}-p1.png"
    ).read_bytes() == PNG_MAGIC + b"NEW-0", (
        "p1's replacement bytes were durable before the failure (atomic "
        "replace) and were NOT removed by the cleanup"
    )
    for n in range(2, 11):
        assert (
            page_dir / f"{doc1.id}-p{n}.png"
        ).read_bytes() == prior[n], f"prior copy p{n} was disturbed"
    assert not list(page_dir.glob("*.tmp")), "zero .tmp orphans from this call"
    assert doc2.chunks, "the document is indexed fully (fail-soft)"


def test_page_copies_are_written_atomically_tmp_then_replace(
    db, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    """THE atomic pin (LOCKED A4): the writer NEVER writes a final
    ``<doc-id>-p{N}.png`` path directly — every page's bytes land on
    ``<name>.png.tmp`` and ``os.replace`` moves them onto the final
    name in order (a reader opening the final path never sees truncated
    bytes), and no ``.tmp`` survives the successful call."""
    root = tmp_path / "pgatom"
    root.mkdir()
    _build_ikea_like_pdf(root / "ikea.pdf")
    llm = _PictorialLLM()
    llm.settings = _settings(tmp_path)
    written: list[Path] = []
    original_write = Path.write_bytes

    def write_spy(self: Path, data: bytes) -> int:
        written.append(self)
        return original_write(self, data)

    replaced: list[tuple[str, str]] = []
    real_replace = os.replace

    def replace_spy(src, dst, *args, **kwargs):  # pyright: ignore[reportUnknownParameterType]
        replaced.append((os.fspath(src), os.fspath(dst)))
        return real_replace(src, dst, *args, **kwargs)

    monkeypatch.setattr(Path, "write_bytes", write_spy)  # pyright: ignore[reportArgumentType]
    monkeypatch.setattr(importer.os, "replace", replace_spy)
    try:
        asyncio.run(import_sources([root], llm, session=db))
    finally:
        monkeypatch.undo()  # pyright: ignore[reportUnknownMemberType]
    doc = _doc(db, "pgatom", "ikea.pdf")
    page_dir = _pages_dir(tmp_path)
    assert len(written) == 10 and len(replaced) == 10
    assert all(p.name.endswith(".png.tmp") for p in written), (
        "no final path is ever written directly"
    )
    for n, (src, dst) in enumerate(replaced, start=1):
        assert dst == str(page_dir / f"{doc.id}-p{n}.png")
        assert src == f"{dst}.tmp", "the tmp name is <final>.tmp"
    assert not list(page_dir.glob("*.tmp")), "no .tmp survives the success path"


def test_stale_tail_delete_failure_is_best_effort(
    db,
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
    caplog: pytest.LogCaptureFixture,
) -> None:
    """An unreadable STALE-TAIL copy (the write-first tail pass, the
    ``_delete_page_copies`` convention): the failure is logged, never
    raised — the sync completes, the row is stamped with the fresh
    count, and the reachable tail files still go."""
    root = tmp_path / "pgtail"
    root.mkdir()
    pdf = root / "ikea.pdf"
    _build_ikea_like_pdf(pdf)
    llm = _PictorialLLM()
    llm.settings = _settings(tmp_path)
    asyncio.run(import_sources([root], llm, session=db))
    doc1 = _doc(db, "pgtail", "ikea.pdf")
    page_dir = _pages_dir(tmp_path)
    p10 = page_dir / f"{doc1.id}-p10.png"
    real_unlink = Path.unlink

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

    monkeypatch.setattr(Path, "unlink", boom)  # pyright: ignore[reportArgumentType]
    _build_ikea_shrunk(pdf)
    try:
        with caplog.at_level(logging.WARNING, logger="app.importer"):
            s2 = asyncio.run(import_sources([root], llm, session=db))
    finally:
        monkeypatch.undo()  # pyright: ignore[reportUnknownMemberType]
    assert (s2.added, s2.updated, s2.unchanged) == (0, 1, 0)
    doc2 = _doc(db, "pgtail", "ikea.pdf")
    assert doc2.page_count == 6 and doc2.page_strip_mode == "additive"
    assert p10.exists(), "the unreadable tail file survived (never raised)"
    for n in range(7, 10):
        assert not (page_dir / f"{doc1.id}-p{n}.png").exists(), f"p{n} still leaked"
    assert any(
        "could not delete PDF page copy" in m for m in caplog.messages
    ), caplog.messages


# ---------------------------------------------------------------------------
# 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")