brain-of-reese/tests/integration/test_importer.py

"""Integration: phase 125, task 03 — the image-anchor enrichment through
the REAL import pipeline (real compose Postgres, the real
``_describe_or_skip`` → ``describe_image`` seam, the mock vision
client — NO seam patch).

A markdown file referencing an image in the SAME source → the vision
call receives the markdown context (appended after the locked
``DESCRIBE_PROMPT``), the image doc's ``image_context`` is set, its
``content``/``summary`` is ``<context>\\n\\n<description>``, and its
chunks are embedded from the enriched content. A standalone image (no
markdown reference) → ``image_context`` NULL, ``content`` the bare
description, the bare locked prompt — byte-identical to pre-phase-125.

Phase 125, task 04 (the markdown-FIRST case) — a markdown indexed while
its referenced image has no document yet stores the reference on the
markdown doc's ``image_context`` (the ``IMAGE_CONTEXT:`` record); the
image that appears later picks the context up at index time (the
task-03 disk scan) or, when the markdown file is gone from disk by
then, through the post-index enrichment pass (the stored doc rows —
record as the index, stored content as the source). Pinned here: the
record-then-image sync pair (the record stored, not cleared), the
pass-only path (markdown file deleted from disk, ``prune=False``), and
the existing-image case (the markdown's record stays NULL — the
task-03 path owns that case).

The unit-level mechanics (the prefix shape, the scan's combined string,
the seam forwarding, the changed-image re-scan, the record formatter +
parser, the pass's idempotency/fail-soft) are pinned in
``tests/unit/test_image_anchor_context.py``; the task-06 story suite
pins the E2E surface.
"""
from __future__ import annotations

import asyncio
import base64
import hashlib
import logging
import tarfile
from collections.abc import Iterator
from pathlib import Path

import pymupdf as fitz  # PyMuPDF — the temp-PDF fixture builder (phase 126)
import pytest
from sqlalchemy import select

from app.config import Settings
from app.models import Chunk, Document
from app.rag.importer import import_sources
from app.rag.llm import EmbeddingError
from app.rag.pdf import (
    PAGE_SEPARATOR,
    extract_pdf_page_texts,
    extract_pdf_text,
    pictorial_pages,
)
from app.rag.summarizer import DESCRIBE_PROMPT, PICTORIAL_PAGE_MODE
from tests.fakes import FakeEmbedder

#: A real 1×1 transparent PNG (the phase-122 fixture — the importer is
#: content-agnostic, the bytes just need to be plausible image bytes).
PNG_1X1 = base64.b64decode(
    "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAAC0lEQVR4nGP4DwQACfsD/fteaysAAAAASUVORK5CYII="
)

DESCRIPTION = (
    "A network diagram of the homelab VLANs: the core switch, the "
    "router, and three labeled subnets."
)

#: The exact context prefix the task-03 extension appends to the locked
#: prompt (``app.rag.summarizer.describe_image``). Phase 137 (task 02,
#: ASSUMPTION A5) generalized the wording — it now covers BOTH the
#: provenance's document context and the phase-125 markdown reference
#: blocks.
CONTEXT_PREFIX = (
    "\n\nThis image is part of a document. The following context "
    "describes where it comes from and what it is used for:\n"
)


class _MockVisionLLM(FakeEmbedder):
    """The phase-122 mock VISION client: the CHAT model answers the
    multimodal describe call with the fixed :data:`DESCRIPTION` (the
    text part is recorded — ``text_parts``); text (``lite``) calls keep
    the ``FakeEmbedder`` behaviour. The REAL
    ``import_sources`` → ``_describe_or_skip`` → ``describe_image``
    chain runs against it (no seam patch)."""

    DESCRIPTION = DESCRIPTION

    def __init__(self) -> None:
        super().__init__()
        self.chat_models: list[str | None] = []
        self.text_parts: list[str] = []

    async def chat(self, messages, model=None):
        self.chat_calls.append(list(messages))
        self.chat_models.append(model)
        user = next((m["content"] for m in messages if m.get("role") == "user"), "")
        if isinstance(user, list):
            self.text_parts.append(user[0]["text"])
            return self.DESCRIPTION
        first = user.split()
        return "Summary of " + (first[0] if first else "<empty>")


def _image_llm[
    LLM: FakeEmbedder
](tmp_path, llm_cls: type[LLM] = _MockVisionLLM, **kwargs) -> LLM:
    """A fake LLM with the phase-122 image knobs (toggle ON; the image
    dir defaults under *tmp_path*). *llm_cls* may be the mock vision
    client (the default — ``text_parts``/``chat_models`` state).
    Phase 140: ``pdf_pictorial`` defaults to OFF here — the pre-phase-
    140 PDF suites in this file pin the pre-phase content shape (their
    fixture pages are sparse — under the 100-char bar — so the
    default-ON pictorial pass would append illustration blocks and
    move the byte-exact pins); the phase-140 suites pass
    ``pdf_pictorial=True`` explicitly."""
    kwargs.setdefault("_env_file", None)
    kwargs.setdefault("images", True)
    kwargs.setdefault("image_dir", str(tmp_path / "images"))
    kwargs.setdefault("pdf_pictorial", False)
    llm = llm_cls()
    llm.settings = Settings(**kwargs)  # pyright: ignore[reportCallIssue]
    return llm


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


#: A markdown file that references ``pic.png`` on line 3 — one context
#: paragraph before and one after (the task-02 window).
GUIDE_MD = (
    "The restore order diagram explains the backup procedure.\n"
    "\n"
    "![Restore order](pic.png)\n"
    "\n"
    "Only run the restore script after the checks pass.\n"
)


def _expected_block(source: str) -> str:
    """The task-03-shape combined string *GUIDE_MD* yields for
    ``<source>/guide.md`` (the disk scan AND the task-04 pass build
    exactly this)."""
    return (
        f"Referenced in {source}/guide.md:\n"
        '- Alt text: "Restore order"\n'
        "  Context: The restore order diagram explains the backup procedure.\n"
        "\n"
        "Only run the restore script after the checks pass."
    )


#: The ``IMAGE_CONTEXT:`` record the markdown-side bookkeeping (phase
#: 125, task 04, work 1) stores for *GUIDE_MD* — the reference on line
#: 3, 2 context paragraphs.
RECORD = (
    "IMAGE_CONTEXT:\n"
    '- [pic.png] (line 3): "Restore order" — 2 paragraphs of context'
)


def test_markdown_referenced_image_gets_context_and_enriched_chunks(
    db, tmp_path
) -> None:
    """A markdown file referencing an image in the same source → sync
    with ``images=True`` (the REAL seam — no patch) → the vision call
    carries the context after the locked prompt, the image doc's
    ``image_context`` is set, its ``content`` includes the context
    (context first, then the description), and its chunks are embedded
    from the enriched content (the ``is_summary`` chunk mirrors it)."""
    source_root = tmp_path / "anchoredkb"
    source_root.mkdir()
    (source_root / "guide.md").write_text(
        "The backup procedure diagram below shows the restore order.\n"
        "\n"
        "![Backup procedure diagram](diagram.png)\n"
        "\n"
        "Run the restore script only after the diagram's checks pass.\n",
        encoding="utf-8",
    )
    (source_root / "diagram.png").write_bytes(PNG_1X1)
    llm = _image_llm(tmp_path)
    try:
        summary = asyncio.run(import_sources([source_root], llm, session=db))
        assert (summary.files, summary.added, summary.images_failed) == (2, 2, 0)

        doc = db.scalar(
            select(Document).where(
                Document.source == source_root.name, Document.path == "diagram.png"
            )
        )
        assert doc is not None and doc.is_image is True and doc.image_path is not None

        expected_context = (
            f"Referenced in {source_root.name}/guide.md:\n"
            '- Alt text: "Backup procedure diagram"\n'
            "  Context: The backup procedure diagram below shows the restore order.\n"
            "\n"
            "Run the restore script only after the diagram's checks pass."
        )
        # The vision model received the context — appended AFTER the
        # locked prompt (the ``IMAGE_DESCRIPTION_MODE`` marker is
        # untouched). The describe call (the ONLY multimodal one) went
        # to the CHAT model (``chat_models`` also carries the markdown
        # doc's ``lite`` summary call — the phase-30 path).
        assert llm.chat_models[0] == llm.settings.llm_chat_model
        assert llm.text_parts == [f"{DESCRIBE_PROMPT}{CONTEXT_PREFIX}{expected_context}"]

        # ``image_context`` stores the combined context (task 05's
        # retrieval display reads from here).
        assert doc.image_context == expected_context
        # The enriched content — context FIRST, then the description —
        # IS the summary (the phase-122 convention) and the chunks are
        # embedded from it (768 dims, the summary chunk mirrors it). The
        # plain-text packer joins the enriched paragraphs with single
        # newlines (the chunker's normalization, PLAN §5), so the content
        # chunk carries the packed form and the summary chunk the raw
        # text.
        expected_content = f"{expected_context}\n\n{DESCRIPTION}"
        assert doc.content == expected_content
        assert doc.summary == expected_content
        packed_content = "\n".join(expected_content.split("\n\n"))
        chunks = db.scalars(select(Chunk).where(Chunk.document_id == doc.id)).all()
        content_chunks = [c for c in chunks if not c.is_summary]
        assert [c.content for c in content_chunks] == [packed_content]
        summary_chunks = [c for c in chunks if c.is_summary]
        assert len(summary_chunks) == 1 and summary_chunks[0].position == -1
        assert summary_chunks[0].content == expected_content
        # Both enriched forms reached the embedder (the markdown doc's
        # two texts complete the four embedded strings).
        embedded = [t for batch in llm.calls for t in batch]
        assert len(embedded) == 4
        assert packed_content in embedded and expected_content in embedded
        for c in chunks:
            assert c.embedding is not None and len(c.embedding) == 768

        # The referencing markdown stays a plain text doc (the
        # ``image_context`` column is an image-doc field — NULL).
        md = db.scalar(
            select(Document).where(
                Document.source == source_root.name, Document.path == "guide.md"
            )
        )
        assert md is not None and md.is_image is False and md.image_context is None
    finally:
        _cleanup_source(db, source_root.name)


def test_standalone_image_keeps_null_context_and_bare_description(db, tmp_path) -> None:
    """A standalone image (no markdown reference in the source — a
    sibling markdown exists but references nothing) → ``image_context``
    NULL, ``content`` the bare description, the vision message the bare
    locked prompt — byte-identical to pre-phase-125 (the chunk set and
    the embedded text are exactly the phase-122 shape)."""
    source_root = tmp_path / "standalonekb"
    source_root.mkdir()
    (source_root / "notes.md").write_text("# Notes\n\nBody text.\n", encoding="utf-8")
    (source_root / "diagram.png").write_bytes(PNG_1X1)
    llm = _image_llm(tmp_path)
    try:
        summary = asyncio.run(import_sources([source_root], llm, session=db))
        assert (summary.files, summary.added, summary.images_failed) == (2, 2, 0)

        doc = db.scalar(
            select(Document).where(
                Document.source == source_root.name, Document.path == "diagram.png"
            )
        )
        assert doc is not None and doc.is_image is True
        assert doc.image_context is None, "standalone: the context stays NULL"
        assert doc.content == DESCRIPTION, "standalone: the bare description"
        assert doc.summary == DESCRIPTION
        assert llm.text_parts == [DESCRIBE_PROMPT], "standalone: the bare prompt"
        chunks = db.scalars(select(Chunk).where(Chunk.document_id == doc.id)).all()
        assert all(c.content == DESCRIPTION for c in chunks)
        for c in chunks:
            assert c.embedding is not None and len(c.embedding) == 768
    finally:
        _cleanup_source(db, source_root.name)


# ---------------------------------------------------------------------------
# Phase 125, task 04 — the markdown-FIRST case: the ``IMAGE_CONTEXT:``
# record on the markdown doc + the post-index enrichment pass.
# ---------------------------------------------------------------------------


def test_markdown_indexed_before_image_stores_record_and_image_picks_it_up(
    db, tmp_path
) -> None:
    """Task 04's primary scenario — a markdown is indexed while the
    image it references does not yet exist: the reference is stored on
    the markdown doc's ``image_context`` (the ``IMAGE_CONTEXT:``
    record). The image file then appears and is synced: the image doc
    is enriched with the markdown's context (the task-03 disk scan —
    the markdown is still on disk), and the markdown's record is NOT
    cleared (ASSUMPTION 4)."""
    source_root = tmp_path / "beforekb"
    source_root.mkdir()
    (source_root / "guide.md").write_text(GUIDE_MD, encoding="utf-8")
    llm = _image_llm(tmp_path)
    try:
        # Run 1 — the markdown alone: the record is stored, no image
        # doc exists.
        s1 = asyncio.run(import_sources([source_root], llm, session=db))
        assert (s1.files, s1.added) == (1, 1)
        md = db.scalar(
            select(Document).where(
                Document.source == source_root.name, Document.path == "guide.md"
            )
        )
        assert md is not None and md.image_context == RECORD
        assert (
            db.scalar(
                select(Document).where(
                    Document.source == source_root.name, Document.path == "pic.png"
                )
            )
            is None
        )

        # Run 2 — the image appears: enriched at index time (the
        # task-03 disk scan sees the on-disk markdown), the record
        # survives.
        (source_root / "pic.png").write_bytes(PNG_1X1)
        s2 = asyncio.run(import_sources([source_root], llm, session=db))
        assert (s2.added, s2.unchanged) == (1, 1)
        block = _expected_block(source_root.name)
        doc = db.scalar(
            select(Document).where(
                Document.source == source_root.name, Document.path == "pic.png"
            )
        )
        assert doc is not None and doc.is_image is True
        assert doc.image_context == block
        expected_content = f"{block}\n\n{DESCRIPTION}"
        assert doc.content == expected_content
        assert doc.summary == expected_content
        # The describe call carried the context after the locked prompt.
        assert llm.text_parts == [f"{DESCRIBE_PROMPT}{CONTEXT_PREFIX}{block}"]
        md2 = db.scalar(
            select(Document).where(
                Document.source == source_root.name, Document.path == "guide.md"
            )
        )
        assert md2 is not None and md2.image_context == RECORD, "the record is NOT cleared"
    finally:
        _cleanup_source(db, source_root.name)


def test_enrichment_pass_enriches_when_markdown_file_is_gone_from_disk(
    db, tmp_path
) -> None:
    """The post-index enrichment pass (task 04, work 3) — the case the
    task-03 disk scan CANNOT cover: the markdown is indexed first (the
    record lands on its doc row), its file is then removed from disk
    (``prune=False`` keeps the row), and only then does the image
    appear. The image is indexed BARE (the disk scan finds no
    reference), and the pass enriches it from the markdown doc's
    STORED rows: the record identifies the referencing markdown, the
    reference is re-extracted from the stored content, and the
    re-describe goes out WITH the context (a second vision call — the
    first, at index time, was bare)."""
    source_root = tmp_path / "passkb"
    source_root.mkdir()
    (source_root / "guide.md").write_text(GUIDE_MD, encoding="utf-8")
    llm = _image_llm(tmp_path)
    try:
        s1 = asyncio.run(import_sources([source_root], llm, session=db))
        assert s1.added == 1
        md = db.scalar(
            select(Document).where(
                Document.source == source_root.name, Document.path == "guide.md"
            )
        )
        assert md is not None and md.image_context == RECORD

        # The markdown file disappears from disk (``prune=False`` keeps
        # the row); the image appears.
        (source_root / "guide.md").unlink()
        (source_root / "pic.png").write_bytes(PNG_1X1)
        s2 = asyncio.run(import_sources([source_root], llm, session=db))
        # The image is added (bare) and then enriched by the pass
        # (updated) — both in this run.
        assert (s2.files, s2.added, s2.updated) == (1, 1, 1)
        block = _expected_block(source_root.name)
        doc = db.scalar(
            select(Document).where(
                Document.source == source_root.name, Document.path == "pic.png"
            )
        )
        assert doc is not None and doc.is_image is True
        assert doc.image_context == block
        expected_content = f"{block}\n\n{DESCRIPTION}"
        assert doc.content == expected_content
        assert doc.summary == expected_content
        # The re-embedded pipeline (the pass replaced the bare chunks).
        packed = "\n".join(expected_content.split("\n\n"))
        chunks = db.scalars(select(Chunk).where(Chunk.document_id == doc.id)).all()
        assert [c.content for c in chunks if not c.is_summary] == [packed]
        assert [c.content for c in chunks if c.is_summary] == [expected_content]
        # Two vision calls: the bare index-time describe (the disk scan
        # saw no reference) and the pass's context-carrying re-describe.
        assert llm.text_parts == [DESCRIBE_PROMPT, f"{DESCRIBE_PROMPT}{CONTEXT_PREFIX}{block}"]
        # The markdown doc's record survives (ASSUMPTION 4).
        md2 = db.scalar(
            select(Document).where(
                Document.source == source_root.name, Document.path == "guide.md"
            )
        )
        assert md2 is not None and md2.image_context == RECORD
    finally:
        _cleanup_source(db, source_root.name)


def test_markdown_referencing_existing_image_keeps_null_record(db, tmp_path) -> None:
    """An EXISTING image doc (indexed before the markdown ever
    existed): the markdown's record stays NULL when it is indexed (the
    task-03 path owns that case — enrichment at IMAGE-index time), the
    unchanged image is not re-scanned in that run, and the image picks
    up the context on its NEXT re-index (bytes changed) via the
    task-03 disk scan — the markdown's record is NULL throughout."""
    source_root = tmp_path / "existingkb"
    source_root.mkdir()
    (source_root / "pic.png").write_bytes(PNG_1X1)
    llm = _image_llm(tmp_path)
    try:
        s1 = asyncio.run(import_sources([source_root], llm, session=db))
        assert s1.added == 1

        # Run 2 — the markdown appears (the image is unchanged).
        (source_root / "guide.md").write_text(GUIDE_MD, encoding="utf-8")
        s2 = asyncio.run(import_sources([source_root], llm, session=db))
        assert (s2.added, s2.unchanged) == (1, 1)
        md = db.scalar(
            select(Document).where(
                Document.source == source_root.name, Document.path == "guide.md"
            )
        )
        assert md is not None and md.image_context is None
        doc = db.scalar(
            select(Document).where(
                Document.source == source_root.name, Document.path == "pic.png"
            )
        )
        assert doc is not None and doc.image_context is None and doc.content == DESCRIPTION

        # Run 3 — the image changes: the re-index re-scans the on-disk
        # markdown (task 03) and enriches; the record is still NULL.
        (source_root / "pic.png").write_bytes(PNG_1X1 + b"changed")
        s3 = asyncio.run(import_sources([source_root], llm, session=db))
        assert (s3.added, s3.updated, s3.unchanged) == (0, 1, 1)
        block = _expected_block(source_root.name)
        doc3 = db.scalar(
            select(Document).where(
                Document.source == source_root.name, Document.path == "pic.png"
            )
        )
        assert doc3 is not None and doc3.image_context == block
        assert doc3.content == f"{block}\n\n{DESCRIPTION}"
        md3 = db.scalar(
            select(Document).where(
                Document.source == source_root.name, Document.path == "guide.md"
            )
        )
        assert md3 is not None and md3.image_context is None
    finally:
        _cleanup_source(db, source_root.name)


# ---------------------------------------------------------------------------
# Phase 126 — PDF indexing through the REAL import pipeline (the real
# ``app.rag.pdf`` extractor, the page-aware chunker, the REAL
# ``_describe_or_skip`` → ``describe_image`` seam — NO seam patch).
# ---------------------------------------------------------------------------

PDF_TITLE = "Quarterly Homelab Report"

PAGE_ONE = "The core switch firmware was upgraded to the stable channel."
PAGE_TWO = "The backup schedule runs nightly at 02:00 and keeps seven rotations."


def _png_bytes(width: int, height: int) -> bytes:
    pix = fitz.Pixmap(fitz.csRGB, fitz.IRect(0, 0, width, height), 0)
    pix.clear_with(200)
    return pix.tobytes("png")


def _build_pdf(
    path, pages: list[tuple[str, list[bytes]]], title: str | None = None
) -> None:
    """One temp PDF (built with PyMuPDF — no binary fixtures): *pages* is
    (page text, embedded-image bytes) per page; *title* sets the metadata
    title (``None`` = leave it empty)."""
    doc = fitz.open()
    for text, imgs in pages:
        page = doc.new_page()
        if text:
            page.insert_text((72, 72), text)
        y = 150
        for data in imgs:
            pix = fitz.Pixmap(data)
            page.insert_image(
                fitz.Rect(72, y, 72 + pix.width, y + pix.height), stream=data
            )
            y += pix.height + 10
    if title is not None:
        doc.set_metadata({"title": title})
    doc.save(path)
    doc.close()


def test_pdf_becomes_document_with_text_chunks_embedding_and_summary(
    db, tmp_path
) -> None:
    """The phase-126 primary scenario: a PDF in a source (the ``pdf``
    import token) → sync → the document carries the extracted text
    (the ``---PAGE-N---`` markers, A2 — the title is the metadata
    title), page-anchored chunks with 768-dim embeddings, and the
    phase-30/118 lite summary. With ``images=True`` (A3) the embedded
    image joins the pipeline through the REAL seam — one vision call on
    the CHAT model with the locked prompt + the phase-137 document
    context (the provenance's origin line + page 2's text — no markdown
    reference exists, so no ``Referenced in`` block), the copy in
    ``image_dir``, and the description (enriched with the context)
    as the image doc's content AND summary."""
    source_root = tmp_path / "pdfkb"
    source_root.mkdir()
    png = _png_bytes(40, 30)
    _build_pdf(
        source_root / "report.pdf",
        [(PAGE_ONE, []), (PAGE_TWO, [png])],
        title=PDF_TITLE,
    )
    llm = _image_llm(tmp_path, import_extensions="md,txt,pdf")
    try:
        summary = asyncio.run(import_sources([source_root], llm, session=db))
        assert (summary.files, summary.added, summary.images_failed) == (1, 2, 0)
        assert summary.formats == {"pdf": 1}

        doc = db.scalar(
            select(Document).where(
                Document.source == source_root.name, Document.path == "report.pdf"
            )
        )
        assert doc is not None and doc.is_image is False
        assert doc.title == PDF_TITLE
        assert doc.content == f"---PAGE-1---\n{PAGE_ONE}\n\n---PAGE-2---\n{PAGE_TWO}"
        chunks = db.scalars(
            select(Chunk).where(Chunk.document_id == doc.id).order_by(Chunk.position)
        ).all()
        content_chunks = [c for c in chunks if not c.is_summary]
        assert [c.content for c in content_chunks] == [
            f"Page 1:\n{PAGE_ONE}",
            f"Page 2:\n{PAGE_TWO}",
        ]
        for c in chunks:
            assert c.embedding is not None and len(c.embedding) == 768
        summary_chunks = [c for c in chunks if c.is_summary]
        assert len(summary_chunks) == 1 and summary_chunks[0].position == -1
        assert doc.summary is not None
        assert doc.summary.endswith(f"Source: {source_root.name}/report.pdf")

        # The embedded image (A3) — through the REAL describe seam.
        image_rel = "report.pdf#page=2#img=0.png"
        img = db.scalar(
            select(Document).where(
                Document.source == source_root.name, Document.path == image_rel
            )
        )
        # Phase 137 (task 02): the provenance's document context — the
        # origin line + page 2's text (no markdown reference → no
        # ``Referenced in`` block) — is stored on ``image_context`` and
        # enriches the content (context first, then the description).
        expected_context = (
            f"Extracted from {source_root.name}/report.pdf (page 2):\n{PAGE_TWO}"
        )
        assert img is not None and img.is_image is True
        assert img.image_context == expected_context
        expected_content = f"{expected_context}\n\n{DESCRIPTION}"
        assert img.content == expected_content and img.summary == expected_content
        copy = Path(img.image_path)
        assert copy.parent == Path(llm.settings.image_dir).expanduser()
        assert copy.name == f"{img.id}.png"
        assert copy.read_bytes() == png
        # The vision call went to the CHAT model with the locked prompt
        # + the generalized prefix + the document context (the PDF's
        # lite summary is the other chat call).
        assert llm.chat_models == [
            llm.settings.llm_summary_model, llm.settings.llm_chat_model
        ]
        assert llm.text_parts == [DESCRIBE_PROMPT + CONTEXT_PREFIX + expected_context]
    finally:
        _cleanup_source(db, source_root.name)


def test_pdf_embedded_images_skipped_when_images_off(db, tmp_path) -> None:
    """With ``images=False`` (the DEFAULT): the PDF text is indexed
    normally and the embedded images are skipped — no image documents,
    no ``images_failed`` (a skip is not a failure), no copies, no vision
    calls."""
    source_root = tmp_path / "pdfoffkb"
    source_root.mkdir()
    _build_pdf(
        source_root / "report.pdf",
        [(PAGE_ONE, [_png_bytes(40, 30)]), (PAGE_TWO, [])],
    )
    llm = _image_llm(tmp_path, images=False, import_extensions="md,txt,pdf")
    try:
        summary = asyncio.run(import_sources([source_root], llm, session=db))
        assert (summary.files, summary.added, summary.images_failed) == (1, 1, 0)
        doc = db.scalar(
            select(Document).where(
                Document.source == source_root.name, Document.path == "report.pdf"
            )
        )
        assert doc is not None and PAGE_ONE in doc.content
        chunks = db.scalars(
            select(Chunk).where(Chunk.document_id == doc.id).order_by(Chunk.position)
        ).all()
        assert any(c.embedding is not None for c in chunks)
        assert (
            db.scalar(
                select(Document).where(
                    Document.source == source_root.name,
                    Document.is_image.is_(True),
                )
            )
            is None
        ), "images off: no embedded-image documents"
        # No vision call at all — only the PDF's lite summary.
        assert llm.chat_models == [llm.settings.llm_summary_model]
    finally:
        _cleanup_source(db, source_root.name)


def test_pdf_without_embedded_images_and_unchanged_rerun(db, tmp_path) -> None:
    """A PDF without embedded images → no image documents; an unchanged
    re-sync → ``unchanged`` (no re-extract, no re-embed, no vision or
    embed calls), and a second run never creates image documents out of
    nothing."""
    source_root = tmp_path / "pdfnonekb"
    source_root.mkdir()
    _build_pdf(
        source_root / "notes.pdf",
        [(PAGE_ONE, []), (PAGE_TWO, [])],
    )
    llm = _image_llm(tmp_path, import_extensions="md,txt,pdf")
    try:
        s1 = asyncio.run(import_sources([source_root], llm, session=db))
        assert (s1.files, s1.added, s1.unchanged) == (1, 1, 0)
        assert (
            db.scalar(
                select(Document).where(
                    Document.source == source_root.name,
                    Document.is_image.is_(True),
                )
            )
            is None
        )
        calls1, chat1 = len(llm.calls), len(llm.chat_models)
        s2 = asyncio.run(import_sources([source_root], llm, session=db))
        assert (s2.files, s2.added, s2.updated, s2.unchanged) == (1, 0, 0, 1)
        assert len(llm.calls) == calls1 and len(llm.chat_models) == chat1
    finally:
        _cleanup_source(db, source_root.name)


def test_pdf_without_text_takes_the_screenshot_fallback(db, tmp_path) -> None:
    """Phase 133 (task 03): a PDF with no extractable text layer
    (image-only, ``images`` OFF — the embedded image is invisible to
    the phase-126 A3 pass, but the FALLBACK is not gated by the
    toggle) → the gate's empty-layer target fires: the single page is
    rendered to a PNG and vision-described (the mock's fixed
    :data:`DESCRIPTION`), and the document lands with the
    ``---PAGE-N---``-anchored description as content (one page-anchored
    content chunk + the ``lite`` summary chunk, both embedded). The
    phase-126 empty-content behavior is retired for layers that CAN be
    screenshotted — it survives only for an unopenable file, where the
    render fails too (the unit suite pins it)."""
    source_root = tmp_path / "pdfemptykb"
    source_root.mkdir()
    _build_pdf(source_root / "pics.pdf", [("", [_png_bytes(40, 30)])])
    llm = _image_llm(tmp_path, images=False, import_extensions="md,txt,pdf")
    try:
        summary = asyncio.run(import_sources([source_root], llm, session=db))
        assert (summary.files, summary.added, summary.images_failed) == (1, 1, 0)
        assert summary.pdfs_screenshotted == 1

        doc = db.scalar(
            select(Document).where(
                Document.source == source_root.name, Document.path == "pics.pdf"
            )
        )
        assert doc is not None and doc.is_image is False
        assert doc.content == f"---PAGE-1---\n{DESCRIPTION}"
        assert doc.content_hash == hashlib.sha256(doc.content.encode("utf-8")).hexdigest()
        chunks = db.scalars(
            select(Chunk).where(Chunk.document_id == doc.id).order_by(Chunk.position)
        ).all()
        content_chunks = [c for c in chunks if not c.is_summary]
        assert [c.content for c in content_chunks] == [f"Page 1:\n{DESCRIPTION}"], (
            "the page-anchored fallback content is chunked like any PDF"
        )
        # The phase-30/118 path runs on the fallback content — the
        # summary chunk mirrors the stored summary, both embedded.
        assert len(chunks) == 2
        assert all(c.embedding is not None and len(c.embedding) == 768 for c in chunks)
        assert doc.summary is not None
    finally:
        _cleanup_source(db, source_root.name)


def test_pdf_embedded_image_survives_prune_and_leaves_with_the_pdf(
    db, tmp_path
) -> None:
    """Phase 126 (A3) prune parent rule — the defect the E2E surfaced:
    the embedded image doc's synthetic path NEVER appears in the walk's
    ``seen``, so the plain prune rule destroyed it in the very sync that
    (re)indexed it (the toggle-on half of the phase-122 "invisible, not
    deleted" guard was missing). Now: with the parent PDF's rel still in
    the walk — even UNCHANGED, so no image pass runs at all — the image
    doc survives (0 pruned); when the PDF LEAVES the source, the image
    doc is pruned with it (row + ``image_dir`` copy — no orphans)."""
    source_root = tmp_path / "pdfprune"
    source_root.mkdir()
    png = _png_bytes(40, 30)
    _build_pdf(
        source_root / "report.pdf",
        [(PAGE_ONE, []), (PAGE_TWO, [png])],
    )
    image_rel = "report.pdf#page=2#img=0.png"
    llm = _image_llm(tmp_path, import_extensions="md,txt,pdf")
    try:
        s1 = asyncio.run(
            import_sources([source_root], llm, session=db, prune=True)
        )
        assert (s1.added, s1.pruned) == (2, 0)
        img1 = db.scalar(
            select(Document).where(
                Document.source == source_root.name, Document.path == image_rel
            )
        )
        assert img1 is not None and img1.image_path is not None
        copy = Path(img1.image_path)

        # Unchanged re-sync (prune ON): the PDF's text digest matches →
        # no image pass at all — the image doc survives anyway (the
        # parent's rel is in the walk's seen set).
        s2 = asyncio.run(
            import_sources([source_root], llm, session=db, prune=True)
        )
        assert (s2.unchanged, s2.pruned) == (1, 0)
        assert db.scalar(
            select(Document).where(
                Document.source == source_root.name, Document.path == image_rel
            )
        ) is not None, "parent PDF present → the image doc survives the prune"

        # The PDF LEAVES the source: the image doc is pruned WITH it
        # (row + copy — no orphan in image_dir).
        (source_root / "report.pdf").unlink()
        s3 = asyncio.run(
            import_sources([source_root], llm, session=db, prune=True)
        )
        assert (s3.pruned, s3.unchanged) == (2, 0)
        assert db.scalar(
            select(Document).where(
                Document.source == source_root.name, Document.path == image_rel
            )
        ) is None, "parent PDF gone → the image doc is pruned with it"
        assert not copy.exists(), "the pruned image's copy leaves image_dir"
    finally:
        _cleanup_source(db, source_root.name)


# --- Phase 125 parity: a markdown file referencing a PDF-embedded image ---

#: A markdown file referencing the PDF's embedded image by its SYNTHETIC
#: path (``<pdf path>#page=<N>#img=<index>.<ext>`` — the doc's ``path``
#: is the upsert key, and the reference resolves to it through the shared
#: ``_resolve_ref_path`` rule) — one context paragraph before, one after.
PDF_GUIDE_MD = (
    "The VLAN map diagram in the report shows the subnet layout.\n"
    "\n"
    "![VLAN map](report.pdf#page=2#img=0.png)\n"
    "\n"
    "Only expand the lab subnet after the VLAN map is reviewed.\n"
)

#: The same reference against ``alpha.pdf`` (the PDF-FIRST ordering's
#: fixture — ``alpha.pdf`` sorts before ``zguide.md`` in the walk).
PDF_ZGUIDE_MD = PDF_GUIDE_MD.replace("report.pdf", "alpha.pdf")


def _pdf_expected_block(source: str, md_name: str, pdf_name: str) -> str:
    """The task-03-shape combined string for the PDF reference (the disk
    scan builds exactly this — the standalone image's shape, the
    synthetic path as the image identity)."""
    return (
        f"Referenced in {source}/{md_name}:\n"
        '- Alt text: "VLAN map"\n'
        "  Context: The VLAN map diagram in the report shows the subnet layout.\n"
        "\n"
        "Only expand the lab subnet after the VLAN map is reviewed."
    )


def test_pdf_embedded_image_referenced_by_markdown_markdown_first(
    db, tmp_path
) -> None:
    """The MARKDOWN-FIRST ordering (``guide.md`` sorts before
    ``report.pdf``): the markdown is indexed while its referenced image
    has no document yet → it records the reference on its OWN row (the
    ``IMAGE_CONTEXT:`` record — phase 125, task 04, work 1, never
    cleared); the PDF is indexed next and the task-03 disk scan (the
    standalone image's parity path — the markdown is still on disk) finds
    the reference, so the image doc is enriched AT INDEX TIME: the vision
    call carries the context after the locked prompt, ``image_context``
    is set, ``content``/``summary`` are ``<context>\\n\\n<description>``.
    The post-index pass then skips the already-enriched doc (idempotent).
    Final state: the record AND the enrichment coexist."""
    source_root = tmp_path / "pdfmdfirst"
    source_root.mkdir()
    (source_root / "guide.md").write_text(PDF_GUIDE_MD, encoding="utf-8")
    _build_pdf(
        source_root / "report.pdf",
        [(PAGE_ONE, []), (PAGE_TWO, [_png_bytes(40, 30)])],
    )
    llm = _image_llm(tmp_path, import_extensions="md,txt,pdf")
    try:
        summary = asyncio.run(import_sources([source_root], llm, session=db))
        # guide.md + report.pdf + the embedded image doc: all added.
        assert (summary.files, summary.added, summary.images_failed) == (2, 3, 0)

        expected_block = _pdf_expected_block(source_root.name, "guide.md", "report.pdf")
        # Phase 137 (task 02, A2): the provenance's document block is
        # FIRST — the combined context the vision call, the stored
        # ``image_context``, and the enriched content all carry.
        expected_context = (
            f"Extracted from {source_root.name}/report.pdf (page 2):\n{PAGE_TWO}"
        ) + "\n\n" + expected_block

        # Work 1 — the markdown doc STORES the reference (the record —
        # line 3 is the reference, 2 context paragraphs). It is the
        # post-index pass's reference index and the retrieval display's
        # record; the pass does not clear it on success.
        md = db.scalar(
            select(Document).where(
                Document.source == source_root.name, Document.path == "guide.md"
            )
        )
        assert md is not None and md.is_image is False
        assert md.image_context == (
            "IMAGE_CONTEXT:\n"
            '- [report.pdf#page=2#img=0.png] (line 3): "VLAN map"'
            " — 2 paragraphs of context"
        )

        # The enrichment (the disk scan at PDF index time) — the image
        # doc's shape is the standalone image's enriched shape.
        image_rel = "report.pdf#page=2#img=0.png"
        img = db.scalar(
            select(Document).where(
                Document.source == source_root.name, Document.path == image_rel
            )
        )
        assert img is not None and img.is_image is True
        assert img.image_context == expected_context
        expected_content = f"{expected_context}\n\n{DESCRIPTION}"
        assert img.content == expected_content
        assert img.summary == expected_content
        # The vision call carried the combined context AFTER the locked
        # prompt (the multimodal part list's text part — the describe
        # seam's phase-125 extension, the phase-137 generalized prefix).
        assert llm.text_parts == [f"{DESCRIBE_PROMPT}{CONTEXT_PREFIX}{expected_context}"]
        # Chat call order (walk order): guide.md's ``lite`` summary,
        # the PDF's ``lite`` summary, then the image's CHAT-model
        # description (the ONLY multimodal call).
        assert llm.chat_models == [
            llm.settings.llm_summary_model,
            llm.settings.llm_summary_model,
            llm.settings.llm_chat_model,
        ]
        # The chunks are embedded from the ENRICHED content (the
        # plain-text packer joins the enriched paragraphs with single
        # newlines; the summary chunk mirrors the raw content).
        packed = "\n".join(expected_content.split("\n\n"))
        chunks = db.scalars(
            select(Chunk).where(Chunk.document_id == img.id).order_by(Chunk.position)
        ).all()
        assert [c.content for c in chunks if not c.is_summary] == [packed]
        summary_chunks = [c for c in chunks if c.is_summary]
        assert len(summary_chunks) == 1 and summary_chunks[0].content == expected_content
        for c in chunks:
            assert c.embedding is not None and len(c.embedding) == 768
    finally:
        _cleanup_source(db, source_root.name)


def test_pdf_embedded_image_referenced_by_markdown_pdf_first(
    db, tmp_path
) -> None:
    """The PDF-FIRST ordering (``alpha.pdf`` sorts before ``zguide.md``
    in the walk): the PDF (and its image doc) is indexed while the
    markdown is still un-indexed — the task-03 disk scan sees the
    reference on disk and enriches the image AT INDEX TIME (the ordering
    that, without the parity scan, would ship an un-enriched image — the
    markdown's record stays NULL: at markdown index time the image doc
    ALREADY exists, so the task-03 path owns the case and nothing is
    recorded). Same final state as the markdown-first test — the
    enrichment never depends on the walk order."""
    source_root = tmp_path / "pdfpdffirst"
    source_root.mkdir()
    (source_root / "zguide.md").write_text(PDF_ZGUIDE_MD, encoding="utf-8")
    _build_pdf(
        source_root / "alpha.pdf",
        [(PAGE_ONE, []), (PAGE_TWO, [_png_bytes(40, 30)])],
    )
    llm = _image_llm(tmp_path, import_extensions="md,txt,pdf")
    try:
        summary = asyncio.run(import_sources([source_root], llm, session=db))
        assert (summary.files, summary.added, summary.images_failed) == (2, 3, 0)

        expected_block = _pdf_expected_block(source_root.name, "zguide.md", "alpha.pdf")
        # Phase 137 (task 02, A2): the provenance's document block is
        # FIRST (the same combined context as the markdown-first final
        # state — the enrichment never depends on the walk order).
        expected_context = (
            f"Extracted from {source_root.name}/alpha.pdf (page 2):\n{PAGE_TWO}"
        ) + "\n\n" + expected_block

        # The task-03 path owns this case: the image doc existed when
        # the markdown was indexed → NO record (the field is the
        # pending-enrichment index, not a reference dump).
        md = db.scalar(
            select(Document).where(
                Document.source == source_root.name, Document.path == "zguide.md"
            )
        )
        assert md is not None and md.image_context is None

        # The image doc is enriched (the disk scan at PDF index time) —
        # byte-identical in shape to the markdown-first final state.
        image_rel = "alpha.pdf#page=2#img=0.png"
        img = db.scalar(
            select(Document).where(
                Document.source == source_root.name, Document.path == image_rel
            )
        )
        assert img is not None and img.image_context == expected_context
        expected_content = f"{expected_context}\n\n{DESCRIPTION}"
        assert img.content == expected_content
        assert img.summary == expected_content
        assert llm.text_parts == [f"{DESCRIBE_PROMPT}{CONTEXT_PREFIX}{expected_context}"]
        # Chat call order (walk order): the PDF's ``lite`` summary, the
        # image's CHAT-model description, then zguide.md's summary.
        assert llm.chat_models == [
            llm.settings.llm_summary_model,
            llm.settings.llm_chat_model,
            llm.settings.llm_summary_model,
        ]
    finally:
        _cleanup_source(db, source_root.name)


# ---------------------------------------------------------------------------
# Phase 137, task 03 — the document-context backfill through the REAL
# import pipeline (seeded pre-phase-137 image rows, the REAL
# ``_backfill_image_contexts`` pass inside ``import_sources``, the REAL
# ``_describe_or_skip`` → ``describe_image`` seam — NO seam patch).
# ---------------------------------------------------------------------------

#: The seeded embedded image's phase-126 synthetic path (the parent
#: ``report.pdf``, page 2, index 0, ``.png`` — the upsert key the
#: forward path would have built).
BACKFILL_IMAGE_REL = "report.pdf#page=2#img=0.png"


def _seed_backfill_row(
    db,
    *,
    source: str,
    image_rel: str,
    image_path: str,
    context: str | None = None,
    title: str = "Embedded image from report page 2",
) -> Document:
    """One pre-phase-137 PDF-embedded image doc row — the shape a
    phase-126 sync left behind: ``image_context`` NULL (or the
    markdown-only phase-125 value), ``content``/``summary`` the bare
    description (or the phase-125 enriched shape), one content chunk +
    the ``is_summary`` chunk, and ``image_path`` naming the persistent
    ``image_dir`` copy (the caller controls whether the file exists).
    """
    content = f"{context}\n\n{DESCRIPTION}" if context is not None else DESCRIPTION
    doc = Document(
        source=source,
        path=image_rel,
        full_path=image_path,
        title=title,
        content=content,
        content_hash=hashlib.sha256(PNG_1X1).hexdigest(),
        is_image=True,
        image_path=image_path,
        image_context=context,
        summary=content,
    )
    db.add(doc)
    db.flush()
    packed = "\n".join(content.split("\n\n"))
    db.add(
        Chunk(
            document_id=doc.id, position=0, content=packed,
            embedding=[0.0] * 768,
        )
    )
    db.add(
        Chunk(
            document_id=doc.id, position=-1, content=content, is_summary=True,
            embedding=[0.0] * 768,
        )
    )
    db.commit()
    return doc


def test_backfill_re_describes_context_free_embedded_image_with_document_context(
    db, tmp_path
) -> None:
    """A pre-phase-137 PDF-embedded image doc (``image_context`` NULL,
    the bare description, the persistent copy) + the parent PDF still
    in the source (with page text) → the next REAL sync backfills:
    exactly ONE describe call whose context carries the document block
    (the origin line + page 2's text); the row's ``image_context``
    starts with ``Extracted from ``, its ``content``/``summary`` is the
    combined context + the fresh description; the chunks are replaced;
    the row's identity fields stay; the log-only counter is 1 (the
    backfill NEVER counts added/updated — the source did not change)."""
    source_root = tmp_path / "bkfillkb"
    source_root.mkdir()
    _build_pdf(source_root / "report.pdf", [(PAGE_ONE, []), (PAGE_TWO, [])])
    copy = tmp_path / "copies" / "seed.png"
    copy.parent.mkdir(parents=True)
    copy.write_bytes(PNG_1X1)
    _seed_backfill_row(
        db, source=source_root.name, image_rel=BACKFILL_IMAGE_REL,
        image_path=str(copy),
    )
    llm = _image_llm(tmp_path, import_extensions="md,txt,pdf")
    try:
        summary = asyncio.run(import_sources([source_root], llm, session=db))
        assert summary.image_context_backfilled == 1
        assert summary.images_failed == 0 and summary.errors == 0
        # The parent PDF is the only WALKED file (added); the backfilled
        # row counts on its own log-only counter, never added/updated.
        assert (summary.files, summary.added, summary.updated) == (1, 1, 0)

        expected_context = (
            f"Extracted from {source_root.name}/report.pdf (page 2):\n{PAGE_TWO}"
        )
        # The ONE describe call carries the document context (the locked
        # prompt + the generalized prefix); the parent's ``lite`` summary
        # is the other chat call, not a describe.
        assert llm.text_parts == [
            f"{DESCRIBE_PROMPT}{CONTEXT_PREFIX}{expected_context}"
        ]
        assert llm.chat_models == [
            llm.settings.llm_summary_model, llm.settings.llm_chat_model
        ]

        img = db.scalar(
            select(Document).where(
                Document.source == source_root.name,
                Document.path == BACKFILL_IMAGE_REL,
            )
        )
        assert img is not None and img.is_image is True
        assert img.image_context == expected_context
        assert img.image_context.startswith("Extracted from ")
        expected_content = f"{expected_context}\n\n{DESCRIPTION}"
        assert img.content == expected_content
        assert img.summary == expected_content
        # The chunks are REPLACED (the bare-description chunk is gone;
        # the enriched content chunk + the mirroring summary chunk are
        # freshly embedded).
        packed = "\n".join(expected_content.split("\n\n"))
        chunks = db.scalars(
            select(Chunk).where(Chunk.document_id == img.id).order_by(Chunk.position)
        ).all()
        assert [c.content for c in chunks if not c.is_summary] == [packed]
        summary_chunks = [c for c in chunks if c.is_summary]
        assert len(summary_chunks) == 1 and summary_chunks[0].position == -1
        assert summary_chunks[0].content == expected_content
        for c in chunks:
            assert c.embedding is not None and len(c.embedding) == 768
        # Deliberately UNCHANGED on the row: the byte digest (the image
        # identity), the copy, the title (the source did not change —
        # the phase-106 date rules do not apply).
        assert img.content_hash == hashlib.sha256(PNG_1X1).hexdigest()
        assert img.image_path == str(copy)
        assert img.title == "Embedded image from report page 2"
        assert copy.read_bytes() == PNG_1X1
    finally:
        _cleanup_source(db, source_root.name)


def test_backfill_is_idempotent_the_next_sync_burns_no_vision_calls(
    db, tmp_path
) -> None:
    """A backfilled doc never qualifies again: the next REAL sync over
    the same source burns ZERO describe calls (the gate no longer
    matches — ``image_context`` starts with ``Extracted from ``), the
    row is byte-identical, and the counter is 0."""
    source_root = tmp_path / "bkfillidempotent"
    source_root.mkdir()
    _build_pdf(source_root / "report.pdf", [(PAGE_ONE, []), (PAGE_TWO, [])])
    copy = tmp_path / "copies" / "seed.png"
    copy.parent.mkdir(parents=True)
    copy.write_bytes(PNG_1X1)
    _seed_backfill_row(
        db, source=source_root.name, image_rel=BACKFILL_IMAGE_REL,
        image_path=str(copy),
    )
    llm = _image_llm(tmp_path, import_extensions="md,txt,pdf")
    try:
        s1 = asyncio.run(import_sources([source_root], llm, session=db))
        assert s1.image_context_backfilled == 1
        img1 = db.scalar(
            select(Document).where(
                Document.source == source_root.name,
                Document.path == BACKFILL_IMAGE_REL,
            )
        )
        assert img1 is not None
        snapshot = (
            img1.content, img1.summary, img1.image_context,
            img1.content_hash, img1.indexed_at, img1.created_at,
            img1.title, img1.image_path,
        )
        describes = len(llm.text_parts)
        assert describes == 1

        # The next sync: the parent PDF is unchanged, the image doc no
        # longer matches the gate.
        s2 = asyncio.run(import_sources([source_root], llm, session=db))
        assert s2.image_context_backfilled == 0
        assert (s2.added, s2.updated, s2.unchanged) == (0, 0, 1)
        assert len(llm.text_parts) == describes  # zero NEW describe calls
        assert s2.images_failed == 0 and s2.errors == 0

        img2 = db.scalar(
            select(Document).where(
                Document.source == source_root.name,
                Document.path == BACKFILL_IMAGE_REL,
            )
        )
        assert img2 is not None
        assert (
            img2.content, img2.summary, img2.image_context,
            img2.content_hash, img2.indexed_at, img2.created_at,
            img2.title, img2.image_path,
        ) == snapshot
    finally:
        _cleanup_source(db, source_root.name)


def test_backfill_parent_gone_gets_the_origin_line_only(db, tmp_path) -> None:
    """The parent PDF is gone from the source dir (no disk dependency):
    the describe call's context is the origin line ONLY (no page-text
    line), and the row is still backfilled — it succeeds once, so the
    gate matches again next sync iff it had failed."""
    source_root = tmp_path / "bkfillgone"
    source_root.mkdir()  # the parent PDF is already gone
    copy = tmp_path / "copies" / "seed.png"
    copy.parent.mkdir(parents=True)
    copy.write_bytes(PNG_1X1)
    _seed_backfill_row(
        db, source=source_root.name, image_rel=BACKFILL_IMAGE_REL,
        image_path=str(copy),
    )
    llm = _image_llm(tmp_path, import_extensions="md,txt,pdf")
    try:
        summary = asyncio.run(import_sources([source_root], llm, session=db))
        assert summary.image_context_backfilled == 1
        assert summary.images_failed == 0 and summary.errors == 0
        expected_context = (
            f"Extracted from {source_root.name}/report.pdf (page 2):"
        )
        assert llm.text_parts == [
            f"{DESCRIBE_PROMPT}{CONTEXT_PREFIX}{expected_context}"
        ]
        img = db.scalar(
            select(Document).where(
                Document.source == source_root.name,
                Document.path == BACKFILL_IMAGE_REL,
            )
        )
        assert img is not None
        assert img.image_context == expected_context  # origin line only
        assert img.content == f"{expected_context}\n\n{DESCRIPTION}"
        assert img.summary == img.content
    finally:
        _cleanup_source(db, source_root.name)


def test_backfill_keeps_the_markdown_block_verbatim_document_first(
    db, tmp_path
) -> None:
    """A markdown-enriched row (the phase-125 ``Referenced in …``
    value — the document block missing) STILL qualifies: the backfilled
    ``image_context`` is the document block + blank line + the stored
    markdown block VERBATIM (document first — re-scanning the disk for
    the markdown context is NOT this pass's job: the phase-125 pass
    owns it and ran first this sync)."""
    source_root = tmp_path / "bkfillmd"
    source_root.mkdir()
    _build_pdf(source_root / "report.pdf", [(PAGE_ONE, []), (PAGE_TWO, [])])
    copy = tmp_path / "copies" / "seed.png"
    copy.parent.mkdir(parents=True)
    copy.write_bytes(PNG_1X1)
    md_block = (
        f"Referenced in {source_root.name}/doc.md:\n"
        '- Alt text: "x"'
    )
    _seed_backfill_row(
        db, source=source_root.name, image_rel=BACKFILL_IMAGE_REL,
        image_path=str(copy), context=md_block,
    )
    llm = _image_llm(tmp_path, import_extensions="md,txt,pdf")
    try:
        summary = asyncio.run(import_sources([source_root], llm, session=db))
        assert summary.image_context_backfilled == 1
        doc_block = (
            f"Extracted from {source_root.name}/report.pdf (page 2):\n{PAGE_TWO}"
        )
        expected_context = f"{doc_block}\n\n{md_block}"
        assert llm.text_parts == [
            f"{DESCRIBE_PROMPT}{CONTEXT_PREFIX}{expected_context}"
        ]
        img = db.scalar(
            select(Document).where(
                Document.source == source_root.name,
                Document.path == BACKFILL_IMAGE_REL,
            )
        )
        assert img is not None
        # Document block FIRST, the markdown block VERBATIM after it.
        assert img.image_context == expected_context
        assert img.content == f"{expected_context}\n\n{DESCRIPTION}"
        assert img.summary == img.content
    finally:
        _cleanup_source(db, source_root.name)


def test_backfill_never_touches_standalone_and_copyless_rows(db, tmp_path) -> None:
    """A standalone image row (a plain path — no ``#page=``) is NEVER a
    candidate (the pattern gate) even with a live copy + NULL context,
    and an embedded-pattern row whose ``image_dir`` copy is gone is
    skipped (the disk gate): both stay byte-identical across TWO REAL
    syncs, with zero describe calls."""
    source_root = tmp_path / "bkfillstandalone"
    source_root.mkdir()
    copy = tmp_path / "copies" / "pic.png"
    copy.parent.mkdir(parents=True)
    copy.write_bytes(PNG_1X1)
    standalone = _seed_backfill_row(
        db, source=source_root.name, image_rel="pic.png",
        image_path=str(copy), title="pic",
    )
    missing_copy = _seed_backfill_row(
        db, source=source_root.name,
        image_rel="notes.pdf#page=1#img=0.png",
        image_path=str(tmp_path / "copies" / "gone.png"),
        title="Embedded image from notes page 1",
    )
    llm = _image_llm(tmp_path, import_extensions="md,txt,pdf")
    try:
        s1 = asyncio.run(import_sources([source_root], llm, session=db))
        s2 = asyncio.run(import_sources([source_root], llm, session=db))
        for s in (s1, s2):
            assert s.image_context_backfilled == 0
            assert s.images_failed == 0 and s.errors == 0
        assert llm.text_parts == []  # zero describe calls, both runs
        # Phase 184: an import run is now ALLOWED to release the session it
        # walks (the identity-map cadence expunges every object at the source
        # boundary), so the two seeded rows are re-read by key instead of
        # expiring the objects the run detached. Same assertion, one level
        # closer to the database — nothing about the run's output changed.
        for row in (standalone, missing_copy):
            reloaded = db.get(Document, row.id)
            assert reloaded is not None
            assert reloaded.image_context is None
            assert reloaded.content == DESCRIPTION
    finally:
        _cleanup_source(db, source_root.name)


def test_backfill_unreadable_copy_is_skipped_without_touching_the_row(
    db, tmp_path
) -> None:
    """A candidate whose copy passes the existence gate but is
    UNREADABLE (a ``read_bytes`` OSError between the gate and the read)
    is skipped — the row stays as-is (the next sync retries it), nothing
    is counted as a failure, and no describe call is made."""
    source_root = tmp_path / "bkfillunreadable"
    source_root.mkdir()
    _build_pdf(source_root / "report.pdf", [(PAGE_ONE, []), (PAGE_TWO, [])])
    copy = tmp_path / "copies" / "seed.png"
    copy.parent.mkdir(parents=True)
    copy.write_bytes(PNG_1X1)
    _seed_backfill_row(
        db, source=source_root.name, image_rel=BACKFILL_IMAGE_REL,
        image_path=str(copy),
    )
    llm = _image_llm(tmp_path, import_extensions="md,txt,pdf")
    try:
        copy.chmod(0o000)  # exists but unreadable (the suite runs non-root)
        try:
            summary = asyncio.run(import_sources([source_root], llm, session=db))
        finally:
            copy.chmod(0o644)
        assert summary.image_context_backfilled == 0
        assert summary.images_failed == 0 and summary.errors == 0
        assert llm.text_parts == []  # no describe attempt
        img = db.scalar(
            select(Document).where(
                Document.source == source_root.name,
                Document.path == BACKFILL_IMAGE_REL,
            )
        )
        assert img is not None
        assert img.image_context is None  # untouched
        assert img.content == DESCRIPTION
    finally:
        _cleanup_source(db, source_root.name)


def test_backfill_recognizes_uppercase_pdf_extension(db, tmp_path) -> None:
    """The walk matches extensions case-insensitively, so an on-disk
    ``REPORT.PDF`` produces a synthetic path with the uppercase suffix —
    the backfill's pattern gate must still recognize it (and the parent
    lookup must hit the uppercase file)."""
    source_root = tmp_path / "bkfillupper"
    source_root.mkdir()
    _build_pdf(source_root / "REPORT.PDF", [(PAGE_ONE, []), (PAGE_TWO, [])])
    copy = tmp_path / "copies" / "seed.png"
    copy.parent.mkdir(parents=True)
    copy.write_bytes(PNG_1X1)
    _seed_backfill_row(
        db, source=source_root.name, image_rel="REPORT.PDF#page=2#img=0.png",
        image_path=str(copy),
    )
    llm = _image_llm(tmp_path, import_extensions="md,txt,pdf")
    try:
        summary = asyncio.run(import_sources([source_root], llm, session=db))
        assert summary.image_context_backfilled == 1
        expected_context = (
            f"Extracted from {source_root.name}/REPORT.PDF (page 2):\n{PAGE_TWO}"
        )
        assert llm.text_parts == [
            f"{DESCRIBE_PROMPT}{CONTEXT_PREFIX}{expected_context}"
        ]
        img = db.scalar(
            select(Document).where(
                Document.source == source_root.name,
                Document.path == "REPORT.PDF#page=2#img=0.png",
            )
        )
        assert img is not None and img.image_context == expected_context
    finally:
        _cleanup_source(db, source_root.name)


class _FirstDescribeFailsLLM(_MockVisionLLM):
    """The first multimodal describe reply is EMPTY (``describe_image``
    → ``None`` — the seam's fail-soft input); later calls answer
    normally (the retry-next-sync proof)."""

    def __init__(self) -> None:
        super().__init__()
        self._failures_left = 1

    async def chat(self, messages, model=None):
        user = next((m["content"] for m in messages if m.get("role") == "user"), "")
        if isinstance(user, list) and self._failures_left:
            self._failures_left -= 1
            saved = self.DESCRIPTION
            self.DESCRIPTION = ""  # the empty reply → the seam's None
            try:
                return await super().chat(messages, model)
            finally:
                self.DESCRIPTION = saved
        return await super().chat(messages, model)


def test_backfill_describe_failure_leaves_row_untouched_and_retries_next_sync(
    db, tmp_path
) -> None:
    """A candidate whose describe call returns ``None`` (an empty vision
    reply) leaves the row UNTOUCHED — nothing was written, so there is
    nothing to roll back — ``images_failed`` + the warning, and the next
    REAL sync (the healthy fake) retries and backfills: the fail-soft
    contract."""
    source_root = tmp_path / "bkfillflaky"
    source_root.mkdir()
    _build_pdf(source_root / "report.pdf", [(PAGE_ONE, []), (PAGE_TWO, [])])
    copy = tmp_path / "copies" / "seed.png"
    copy.parent.mkdir(parents=True)
    copy.write_bytes(PNG_1X1)
    _seed_backfill_row(
        db, source=source_root.name, image_rel=BACKFILL_IMAGE_REL,
        image_path=str(copy),
    )
    llm = _image_llm(
        tmp_path, llm_cls=_FirstDescribeFailsLLM, import_extensions="md,txt,pdf"
    )
    try:
        s1 = asyncio.run(import_sources([source_root], llm, session=db))
        assert s1.images_failed == 1
        assert s1.image_context_backfilled == 0 and s1.errors == 0
        # The failed call itself carried the right context.
        assert len(llm.text_parts) == 1
        assert "Extracted from" in llm.text_parts[0]

        img = db.scalar(
            select(Document).where(
                Document.source == source_root.name,
                Document.path == BACKFILL_IMAGE_REL,
            )
        )
        assert img is not None
        assert img.image_context is None  # untouched
        assert img.content == DESCRIPTION
        assert img.summary == DESCRIPTION
        assert img.content_hash == hashlib.sha256(PNG_1X1).hexdigest()

        # The next sync (the fake is healthy now): the gate still
        # matches — the retry succeeds.
        s2 = asyncio.run(import_sources([source_root], llm, session=db))
        assert s2.image_context_backfilled == 1 and s2.images_failed == 0
        img2 = db.scalar(
            select(Document).where(
                Document.source == source_root.name,
                Document.path == BACKFILL_IMAGE_REL,
            )
        )
        assert img2 is not None and img2.image_context is not None
        assert img2.image_context.startswith("Extracted from ")
        assert img2.content == f"{img2.image_context}\n\n{DESCRIPTION}"
    finally:
        _cleanup_source(db, source_root.name)


class _BackfillEmbedFailsLLM(_MockVisionLLM):
    """The first embed of the backfilled content (a text starting with
    the ``Extracted from`` header) raises :class:`EmbeddingError`; later
    embeds succeed (the retry-next-sync proof for the embed path)."""

    def __init__(self) -> None:
        super().__init__()
        self._failures_left = 1

    async def embed(self, texts, *, on_progress=None):
        if self._failures_left and any(
            t.startswith("Extracted from") for t in texts
        ):
            self._failures_left -= 1
            raise EmbeddingError("embed boom (backfill test)")
        return await super().embed(texts, on_progress=on_progress)


def test_backfill_embed_failure_rolls_back_the_image_and_retries_next_sync(
    db, tmp_path
) -> None:
    """A candidate whose re-chunk/re-embed gives up (an
    ``EmbeddingError``) rolls back THIS image only — the row is
    untouched, the copy STAYS, ``errors`` + the error line — and the
    next REAL sync (the healthy fake) retries and backfills."""
    source_root = tmp_path / "bkfillembedfail"
    source_root.mkdir()
    _build_pdf(source_root / "report.pdf", [(PAGE_ONE, []), (PAGE_TWO, [])])
    copy = tmp_path / "copies" / "seed.png"
    copy.parent.mkdir(parents=True)
    copy.write_bytes(PNG_1X1)
    _seed_backfill_row(
        db, source=source_root.name, image_rel=BACKFILL_IMAGE_REL,
        image_path=str(copy),
    )
    llm = _image_llm(
        tmp_path, llm_cls=_BackfillEmbedFailsLLM, import_extensions="md,txt,pdf"
    )
    try:
        s1 = asyncio.run(import_sources([source_root], llm, session=db))
        assert s1.errors == 1
        assert s1.image_context_backfilled == 0 and s1.images_failed == 0
        # The describe SUCCEEDED (the embed is what failed) — one
        # context-bearing call.
        assert len(llm.text_parts) == 1

        img = db.scalar(
            select(Document).where(
                Document.source == source_root.name,
                Document.path == BACKFILL_IMAGE_REL,
            )
        )
        assert img is not None
        assert img.image_context is None  # rolled back
        assert img.content == DESCRIPTION
        assert img.content_hash == hashlib.sha256(PNG_1X1).hexdigest()
        assert copy.read_bytes() == PNG_1X1  # the copy stays

        s2 = asyncio.run(import_sources([source_root], llm, session=db))
        assert s2.image_context_backfilled == 1 and s2.errors == 0
        img2 = db.scalar(
            select(Document).where(
                Document.source == source_root.name,
                Document.path == BACKFILL_IMAGE_REL,
            )
        )
        assert img2 is not None and img2.image_context is not None
        assert img2.image_context.startswith("Extracted from ")
    finally:
        _cleanup_source(db, source_root.name)


# ---------------------------------------------------------------------------
# Phase 140 — Pictorial PDFs: sparse pages rendered + vision-described
# into the indexed content (the owner's IKEA repro, TODO L5; the kill
# switch; the phase-133 precedence pin).
# ---------------------------------------------------------------------------

REPO_ROOT = Path(__file__).resolve().parents[2]

#: The owner's untracked fixture (TODO L5 — the stable reference: the
#: TARBALL, never the extracted PDF; the phase-133 real-fixture
#: convention — ``test_sources/`` is gitignored, the test extracts it
#: into ``tmp_path`` and never touches the repo, and the owner's
#: machine always has it: a missing tar skips LOUDLY with the named
#: reason, it is never silently green).
IKEA_TARBALL = REPO_ROOT / "test_sources" / "ikea_instructions.tar"

#: The temp source dir's basename is the source name (phase 38).
IKEA_SOURCE = "ikea"

#: The fixture's embedded metadata title (Adobe InDesign's
#: "Assembly Instructions" — the phase-140 context).
IKEA_TITLE = "Assembly Instructions"

#: The fixture's text-heavy pages (pages 2–5 — the multilingual
#: warnings, 3,091–7,288 chars — never candidates; the sparse pages
#: run 4–73 chars, so the 100-char bar sits between the classes).
IKEA_TEXT_HEAVY_PAGES = (2, 3, 4, 5)


class _PictorialVisionLLM(FakeEmbedder):
    """The capturing multimodal fake (task 03): a vision call whose
    text part carries ``PICTORIAL_PAGE_MODE`` (the pictorial pass's
    seam — ``describe_pdf_page``) gets the fixed per-candidate canned
    reply and is recorded in ``pictorial_texts``; a call carrying
    ``IMAGE_DESCRIPTION_MODE`` (the phase-133 fallback's / phase-122
    images' seam — ``describe_image``) gets a DISTINCT canned reply
    and is recorded in ``image_texts``. The fixed replies make the
    stored content a PURE FUNCTION of the fixture (the digest-stable
    premise of the re-sync assertion), and the two recordings witness
    WHICH seam fired (the precedence pin). String (``lite``) calls
    keep the ``FakeEmbedder`` behaviour. The REAL
    ``import_sources`` → ``_pictorial_pdf_pass`` →
    ``describe_pdf_page`` chain runs against it (no seam patch)."""

    PICTORIAL_ANSWER = (
        "IKEA pictorial page: the illustration shows two parts being "
        "joined with a blue arrow; a cam lock is tightened with the "
        "included hex key."
    )
    IMAGE_ANSWER = (
        "Fallback screenshot: the page shows an assembly pictogram."
    )

    def __init__(self) -> None:
        super().__init__()
        self.pictorial_texts: list[str] = []
        self.image_texts: list[str] = []

    async def chat(self, messages, model=None):
        self.chat_calls.append(list(messages))
        user = next((m["content"] for m in messages if m.get("role") == "user"), "")
        if isinstance(user, list):
            text = next((p["text"] for p in user if p.get("type") == "text"), "")
            if PICTORIAL_PAGE_MODE in text:
                self.pictorial_texts.append(text)
                return self.PICTORIAL_ANSWER
            self.image_texts.append(text)
            return self.IMAGE_ANSWER
        first = user.split()
        return "Summary of " + (first[0] if first else "<empty>")


@pytest.fixture()
def ikea_pdf(db, tmp_path) -> Iterator[Path]:
    """The owner's IKEA tarball extracted into a temp ``ikea/`` source
    dir (the phase-133 real-fixture convention — ``tmp_path`` only,
    never the repo). Any leftover ``ikea`` source rows from a crashed
    earlier run are cleaned before AND after (the shared dev DB is
    reused across runs). Skips LOUDLY with the named reason when the
    untracked owner fixture is missing."""
    if not IKEA_TARBALL.is_file():
        pytest.skip(
            "test_sources/ikea_instructions.tar is missing — the untracked "
            "owner fixture (one 24-page vector-drawn assembly-instruction "
            "PDF; the owner's machine always has it)"
        )
    _cleanup_source(db, IKEA_SOURCE)
    root = tmp_path / IKEA_SOURCE
    root.mkdir()
    with tarfile.open(IKEA_TARBALL) as tar:
        tar.extractall(root, filter="data")
    pdfs = sorted(root.rglob("*.pdf"))
    assert len(pdfs) == 1, f"the IKEA tar carries exactly one PDF, found: {pdfs}"
    yield pdfs[0]
    _cleanup_source(db, IKEA_SOURCE)


def test_ikea_tar_sparse_pages_are_described_and_rerun_unchanged(
    db, tmp_path, ikea_pdf: Path, caplog: pytest.LogCaptureFixture
) -> None:
    """The owner's EXACT file (TODO L5) through the REAL importer:
    the deterministic gate's candidates — computed IN the test from
    ``extract_pdf_page_texts`` + ``pictorial_pages`` (no hardcoded
    page list — the fixture is stable but the test must not rot) —
    every one carries ``[Page N illustration: <canned>]`` AFTER its
    own text (LOCKED A2 — no LLM judge: every described page joins
    the content; A4 — the page's text stays), the text-heavy pages
    (2–5) carry NO block, the chunks carry the canned description
    (the owner's rationale — more information is better for the
    embeddings), the title is the metadata title, the digest is over
    the FINAL content, and a re-sync with the same fake lands
    ``unchanged`` through the phase-147 FAST PATH (the raw bytes are
    the change signal: byte-stable row, NO re-embed, and the vision
    calls are NOT re-issued — the pass never runs on an unchanged
    file; the pre-phase "vision re-issued before the unchanged check"
    no-cache cost is gone)."""
    pdf_path = ikea_pdf
    llm = _image_llm(
        tmp_path,
        llm_cls=_PictorialVisionLLM,
        pdf_pictorial=True,
        import_extensions="md,txt,pdf",
    )
    try:
        with caplog.at_level(logging.INFO, logger="app.importer"):
            s1 = asyncio.run(import_sources([pdf_path.parent], llm, session=db))
        assert (s1.files, s1.added, s1.errors) == (1, 1, 0)
        assert s1.formats == {"pdf": 1}

        # The candidates — computed IN the test (no hardcoded page
        # list). The fixture class: the sparse pages are 4–73 chars,
        # the text-heavy pages 3,091+ — the 100-char bar separates.
        page_texts = extract_pdf_page_texts(pdf_path)
        candidates = pictorial_pages(
            page_texts, min_chars=llm.settings.pdf_pictorial_min_chars
        )
        assert candidates, "the fixture must carry sparse (pictorial) pages"
        for n in IKEA_TEXT_HEAVY_PAGES:
            assert n not in candidates, f"page {n} is text-heavy, never a candidate"

        doc = db.scalar(
            select(Document).where(
                Document.source == IKEA_SOURCE, Document.path == pdf_path.name
            )
        )
        assert doc is not None, "the IKEA PDF did not import"
        assert doc.title == IKEA_TITLE  # the PDF's metadata title
        assert doc.is_image is False

        content = doc.content
        # Every page's block is byte-faithful: the page text, the
        # described candidates carrying their illustration block
        # AFTER the text (A4), and NO block on the non-candidates.
        for n, text in page_texts.items():
            block = f"{PAGE_SEPARATOR.format(page=n)}\n{text}"
            if n in candidates:
                block += (
                    f"\n\n[Page {n} illustration: "
                    f"{_PictorialVisionLLM.PICTORIAL_ANSWER}]"
                )
            assert block in content, f"page {n}'s block is not byte-faithful"
            if n not in candidates:
                assert f"[Page {n} illustration" not in content
        assert content.count(" illustration: ") == len(candidates)
        # The digest runs over the FINAL content (the phase-126/133
        # convention — A4).
        assert doc.content_hash == hashlib.sha256(
            content.encode("utf-8")
        ).hexdigest()

        # The chunks carry the canned description (the owner's L5
        # rationale — the embeddings hit the illustrations), and the
        # phase-30/118 summary ran on the final content.
        chunks = db.scalars(
            select(Chunk).where(Chunk.document_id == doc.id).order_by(Chunk.position)
        ).all()
        content_chunks = [c for c in chunks if not c.is_summary]
        assert any(
            _PictorialVisionLLM.PICTORIAL_ANSWER in c.content for c in content_chunks
        ), "no content chunk carries the canned description"
        for c in chunks:
            assert c.embedding is not None and len(c.embedding) == 768
        assert doc.summary is not None

        # The counters (log-only — the sync detail shape is untouched)
        # + the ONE loud log line (PLAN §9), with the computed N.
        assert s1.pdfs_pictorial == 1
        assert s1.pdf_pictorial_pages == len(candidates)
        assert s1.pdfs_screenshotted == 0  # the layer is USABLE — no fallback
        assert llm.image_texts == []  # the fallback seam never fired
        min_chars = llm.settings.pdf_pictorial_min_chars
        lines = [
            r.getMessage() for r in caplog.records if "pictorial pass" in r.getMessage()
        ]
        assert lines == [
            f"pdf: pictorial pass source={IKEA_SOURCE} path={pdf_path.name} — "
            f"{len(candidates)} candidate pages, {len(candidates)} described "
            f"(min_chars={min_chars})"
        ]

        # The re-sync with the SAME fake: ``unchanged`` through the
        # phase-147 fast path — the raw bytes are the change signal,
        # so the pass never re-runs: no re-render, the vision calls
        # are NOT re-issued (the pre-phase "vision re-issued before
        # the unchanged check can skip them" no-cache cost is gone),
        # no re-chunk, no re-embed, no re-summary, no KB change.
        embeds1 = len(llm.calls)
        pictorial1 = len(llm.pictorial_texts)
        content1, digest1 = doc.content, doc.content_hash
        s2 = asyncio.run(import_sources([pdf_path.parent], llm, session=db))
        assert (s2.added, s2.updated, s2.unchanged) == (0, 0, 1)
        assert s2.pdfs_pictorial == 0  # the pass did not re-fire
        assert len(llm.pictorial_texts) == pictorial1  # NO re-issued vision
        assert len(llm.calls) == embeds1, "unchanged: no re-embed"
        assert s2.changed_paths == [], "unchanged: no KB change recorded"
        doc2 = db.scalar(
            select(Document).where(
                Document.source == IKEA_SOURCE, Document.path == pdf_path.name
            )
        )
        assert doc2 is not None
        assert doc2.content == content1  # byte-stable content
        assert doc2.content_hash == digest1  # byte-stable digest
    finally:
        _cleanup_source(db, IKEA_SOURCE)


def test_ikea_tar_kill_switch_is_byte_identical_to_extraction(
    db, tmp_path, ikea_pdf: Path
) -> None:
    """``BOR_PDF_PICTORIAL=0`` (the house kill switch — LOCKED A3):
    the pass never runs on the owner's file — no vision call, no
    counters, no log line — and the stored content is
    ``extract_pdf_text``'s output BYTE-IDENTICAL to the pre-phase-140
    path."""
    pdf_path = ikea_pdf
    llm = _image_llm(
        tmp_path,
        llm_cls=_PictorialVisionLLM,
        pdf_pictorial=False,
        import_extensions="md,txt,pdf",
    )
    try:
        s1 = asyncio.run(import_sources([pdf_path.parent], llm, session=db))
        assert (s1.files, s1.added, s1.errors) == (1, 1, 0)
        doc = db.scalar(
            select(Document).where(
                Document.source == IKEA_SOURCE, Document.path == pdf_path.name
            )
        )
        assert doc is not None
        assert doc.content == extract_pdf_text(pdf_path), (
            "the kill switch must keep the pre-phase-140 content "
            "byte-identical"
        )
        assert llm.pictorial_texts == []  # zero vision calls
        assert (s1.pdfs_pictorial, s1.pdf_pictorial_pages) == (0, 0)
    finally:
        _cleanup_source(db, IKEA_SOURCE)


def test_garbage_pdf_with_pictorial_on_takes_the_fallback_not_the_pass(
    db, tmp_path
) -> None:
    """The precedence pin (LOCKED): an UNUSABLE text layer takes the
    phase-133 full-page screenshot fallback — the pictorial pass
    NEVER runs and never double-visions (the gate is the only gate):
    with ``pdf_pictorial=True`` (the code default), a garbage-text
    PDF (the phase-133 synthetic recipe — XML-ish lines) lands with
    the fallback's full-page canned descriptions, and the fake
    records ZERO ``PICTORIAL_PAGE_MODE`` calls."""
    source_root = tmp_path / "garbpike"
    source_root.mkdir()
    doc = fitz.open()
    for p in (1, 2):
        page = doc.new_page()
        for i in range(10):
            page.insert_text((72, 72 + i * 14), f"<foo:p{p}>{i}</foo:p{p}>")
    doc.save(source_root / "garbage.pdf")
    doc.close()
    llm = _image_llm(
        tmp_path,
        llm_cls=_PictorialVisionLLM,
        pdf_pictorial=True,
        import_extensions="md,txt,pdf",
    )
    try:
        summary = asyncio.run(import_sources([source_root], llm, session=db))
        assert (summary.files, summary.added, summary.errors) == (1, 1, 0)
        assert summary.pdfs_screenshotted == 1  # the gate fired → the fallback
        assert summary.pdfs_pictorial == 0
        assert summary.pdf_pictorial_pages == 0
        assert llm.pictorial_texts == [], (
            "the pictorial pass must never run on an unusable layer "
            "(no double vision)"
        )
        assert len(llm.image_texts) == 2  # one fallback describe per page
        row = db.scalar(
            select(Document).where(
                Document.source == source_root.name, Document.path == "garbage.pdf"
            )
        )
        assert row is not None
        assert row.content == (
            f"---PAGE-1---\n{_PictorialVisionLLM.IMAGE_ANSWER}\n\n"
            f"---PAGE-2---\n{_PictorialVisionLLM.IMAGE_ANSWER}"
        )
    finally:
        _cleanup_source(db, source_root.name)