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

"""Unit tests: importer directory walk + sha256 delta logic + summaries.

The walk tests are pure filesystem (``tmp_path``); the delta and summary
tests run against the local compose Postgres (preferred — a real vector
table), skipping with clear instructions when the stack is not up.

Summaries (phase 30; phase 118, A2: every file, markdown included):
every file gets a ``lite``-model summary via the fake's deterministic
``chat`` (``"Summary of <first token>"``); an unchanged doc whose summary
is NULL is backfilled on the next run (``summary_backfilled``); the
sentinel word ``SUMMARY-BLOWUP`` makes ``chat`` raise :class:`LLMError`
for the fail-soft path.
"""
from __future__ import annotations

import asyncio
import base64
import hashlib
import logging
import os
import re
import textwrap
from collections.abc import Callable, Sequence
from datetime import UTC, datetime
from pathlib import Path

import pymupdf as fitz
import pytest
from sqlalchemy import func, select
from sqlalchemy.orm import Session

import app.rag.importer as importer
from app.config import Settings
from app.models import Chunk, Document
from app.rag.chunker import chunk_document
from app.rag.importer import (
    EXCLUDED_DIRS,
    ImportSummary,
    _prune,
    _store_summary,
    import_sources,
    iter_importable_files,
    match_extension,
)
from app.rag.llm import EmbeddingError, LLMError
from app.rag.pdf import (
    extract_pdf_images_with_stats,
    extract_pdf_page_texts,
    pictorial_pages,
)
from tests.fakes import FakeEmbedder

#: The original seven A9 formats as dotted suffixes (pre-phase-47 default
#: set). The phase-47 walk test uses the live config default
#: (``Settings().import_extension_set`` — now seventeen formats).
DEFAULT_EXTS = frozenset({".md", ".markdown", ".txt", ".yaml", ".yml", ".json", ".py"})

#: Phase 187 (task 02): the ONE rendering :func:`app.core.memory.fmt_mb`
#: allows for an MB field — ``?`` when the platform cannot know, else a
#: positive integer (never ``None``, never a float or a negative).
MB = r"(?:\?|[1-9][0-9]*)"


class _PoisonEmbedder(FakeEmbedder):
    """Fails (like a real endpoint) on any text containing 'poison'."""

    async def embed(
        self,
        texts: list[str],
        *,
        on_progress: Callable[[int, int], None] | None = None,
    ) -> list[list[float]]:
        if any("poison" in t for t in texts):
            raise EmbeddingError("embeddings endpoint refused the input (simulated)")
        return await super().embed(texts, on_progress=on_progress)


class _FailingChatEmbedder(FakeEmbedder):
    """A ``lite`` model that always fails (drives the summary fail-soft
    path — including the phase-118 backfill)."""

    async def chat(self, messages: list[dict[str, str]], model: str | None = None) -> str:
        self.chat_calls.append(list(messages))
        raise LLMError("simulated lite-model failure (test sentinel)")


class _CapEmbedder(FakeEmbedder):
    """Simulates the endpoint's ~1024-token input cap at ~1.1 chars/token:
    any single text over 1000 chars is rejected (URL-dense worst case)."""

    async def embed(
        self,
        texts: list[str],
        *,
        on_progress: Callable[[int, int], None] | None = None,
    ) -> list[list[float]]:
        if any(len(t) > 1000 for t in texts):
            raise EmbeddingError(
                "a single 1100-char chunk exceeded the endpoint's per-request "
                "input token cap — lower BOR_CHUNK_TARGET_CHARS and re-import"
            )
        return await super().embed(texts, on_progress=on_progress)


def _embedder_with_extensions(extensions: str) -> FakeEmbedder:
    """A :class:`FakeEmbedder` whose settings carry a custom
    ``BOR_IMPORT_EXTENSIONS`` CSV (phase 102 — the import scope the walk
    and the ``formats`` counter read from ``llm.settings``)."""
    llm = FakeEmbedder()
    llm.settings = Settings(_env_file=None, import_extensions=extensions)  # 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()


def test_iter_importable_files_excludes_noncontent_dirs_and_hidden(tmp_path: Path) -> None:
    """Well-known non-content dirs, hidden (dot-) dirs/files, and non-A9
    extensions are all skipped; the A9 formats pass."""
    root = tmp_path / "proj"
    for d in (
        "notes/sub",
        ".venv/lib",
        "node_modules/x",
        ".git",
        "__pycache__",
        ".pytest_cache",
        "dist",
        "build",
        ".esphome/.espressif",  # vendored hidden cache — the real A9 case
    ):
        (root / d).mkdir(parents=True)
    files = {
        # content that must be found:
        "README.md": "readme",
        "notes/sub/deep.md": "deep",
        "compose.yaml": "services: {}",
        "legacy.YML": "a: b",  # case-insensitive suffix
        "notes/sub/agent.py": "x = 1",
        "config.json": "{}",
        "README.txt": "plain",
        "notes/sub/deep.markdown": "md2",
        # must be skipped:
        ".venv/lib/junk.md": "junk",
        "node_modules/x/j.md": "j",
        ".git/c.md": "g",
        "__pycache__/c.md": "p",
        ".pytest_cache/c.md": "pc",
        "dist/d.md": "d",
        "build/b.md": "b",
        ".esphome/.espressif/secret.md": "vendor",
        ".secret.md": "hidden file",  # dot-prefixed FILE, not just dir
        "notes/sub/notes.csv": "a,b",  # not an A9 format
        "notes/sub/file.md.bak": "x",
    }
    for rel, text in files.items():
        (root / rel).write_text(text)

    found = {p.relative_to(root).as_posix() for p in iter_importable_files(root, DEFAULT_EXTS)}
    assert found == {
        "README.md",
        "notes/sub/deep.md",
        "compose.yaml",
        "legacy.YML",
        "notes/sub/agent.py",
        "config.json",
        "README.txt",
        "notes/sub/deep.markdown",
    }


def test_iter_importable_files_missing_dir_yields_nothing(tmp_path: Path) -> None:
    assert iter_importable_files(tmp_path / "definitely-missing", DEFAULT_EXTS) == []


def test_iter_importable_files_walker_picks_up_new_formats_by_default(
    tmp_path: Path,
) -> None:
    """Phase 47 (A9 revised 2026-08-27): the ten new formats — the full
    quadlet family + j2 — walk under the *default* extension set (no env
    configuration needed), the original seven still walk (regression), and
    unknown/hidden-dir/excluded-dir filtering is unchanged for them."""
    root = tmp_path / "quadproj"
    (root / ".esphome").mkdir(parents=True)
    (root / "node_modules").mkdir()
    files = {
        # the ten new formats (A9 revised 2026-08-27):
        "web.container": "# RESE-QUADLET-SENTINEL-77aa\n[Container]\nImage=alpine\n",
        "lan.network": "[Network]\nDriver=bridge\n",
        "cache.volume": "[Volume]\nDriver=local\n",
        "alpine.image": "[Image]\nImages=alpine\n",
        "app.pod": "[Pod]\nPodName=app\n",
        "k3s.kube": "apiVersion: v1\nkind: Pod\n",
        "zram.swap": "[Swap]\nFile=/swapfile\n",
        "fedora.os": "[OS]\nImage=fedora\n",
        "edge.endpoint": "[Endpoint]\nPort=8080\n",
        "deploy.j2": "{% for s in services %}\n[{{ s }}]\n{% endfor %}\n",
        # the original seven — regression:
        "note.md": "# n\n",
        "note.markdown": "# m\n",
        "note.txt": "t\n",
        "cfg.yaml": "a: b\n",
        "cfg.yml": "a: b\n",
        "data.json": "{}\n",
        "agent.py": "x = 1\n",
        # must be skipped — filtering is unchanged for the new formats too:
        "mystery.xyz": "unknown extension",
        ".esphome/x.container": "hidden dir (new-format file)",
        "node_modules/y.container": "excluded dir (new-format file)",
    }
    for rel, text in files.items():
        (root / rel).write_text(text)
    # The live config default (Settings with no env override) — no
    # BOR_IMPORT_EXTENSIONS needed for the new formats to walk.
    exts = Settings(_env_file=None).import_extension_set  # pyright: ignore[reportCallIssue]
    found = {p.relative_to(root).as_posix() for p in iter_importable_files(root, exts)}
    assert found == {
        # all ten new formats:
        "web.container",
        "lan.network",
        "cache.volume",
        "alpine.image",
        "app.pod",
        "k3s.kube",
        "zram.swap",
        "fedora.os",
        "edge.endpoint",
        "deploy.j2",
        # …and the original seven:
        "note.md",
        "note.markdown",
        "note.txt",
        "cfg.yaml",
        "cfg.yml",
        "data.json",
        "agent.py",
    }


def test_iter_importable_files_respects_custom_extension_filter(tmp_path: Path) -> None:
    """A narrower filter (e.g. md only) excludes the other A9 formats."""
    root = tmp_path / "filtered"
    root.mkdir()
    (root / "a.md").write_text("a")
    (root / "b.yaml").write_text("a: b")
    (root / "c.py").write_text("x = 1")
    found = {p.name for p in iter_importable_files(root, frozenset([".md"]))}
    assert found == {"a.md"}


def test_match_extension_matrix() -> None:
    """Phase 102, D1 — the matching matrix: the A9 dotted-suffix rule first,
    then the extensionless exact-name rule (case-insensitive), exact name
    only — no partial matching, suffixed files governed by their suffix."""
    exts = frozenset({".md", ".dockerfile"})
    # Rule 1 — the dotted suffix (case-insensitive, as today):
    assert match_extension(Path("kubernetes.md"), frozenset({".md"})) == "md"
    assert match_extension(Path("Kubernetes.MD"), frozenset({".md"})) == "md"
    # Rule 2 — extensionless files by exact lowercased FULL filename:
    assert match_extension(Path("Dockerfile"), exts) == "dockerfile"
    assert match_extension(Path("DOCKERFILE"), exts) == "dockerfile"
    # …only when the token is actually in the set:
    assert match_extension(Path("Dockerfile"), frozenset({".md"})) is None
    # Exact name only — compound lookalikes never match the token:
    assert match_extension(Path("mydockerfile"), exts) is None
    # A suffixed file is governed by its suffix, never its name:
    assert match_extension(Path("Dockerfile.dev"), exts) is None
    assert match_extension(Path("Dockerfile.dev"), frozenset({".md", ".dev"})) == "dev"
    # An out-of-scope suffix is out of scope, name be damned:
    assert match_extension(Path("readme.rst"), frozenset({".md"})) is None


def test_iter_importable_files_walks_extensionless_name_tokens(tmp_path: Path) -> None:
    """Phase 102, D1 — an extensionless file walks iff its lowercased full
    name is a token (``md,dockerfile,containerfile`` here): lookalikes and
    the dot-prefixed hidden file stay skipped by the pre-existing rules."""
    root = tmp_path / "build"
    root.mkdir()
    for name in (
        "Dockerfile",
        "Containerfile",
        "mydockerfile",  # compound name — never matches the `dockerfile` token
        "Dockerfile.dev",  # governed by its .dev suffix (not a token here)
        ".dockerfile",  # dot-prefixed FILE — the hidden-component rule
        "notes.md",
    ):
        (root / name).write_text(f"# {name}\n\nbody {name}\n")
    exts = frozenset({".md", ".dockerfile", ".containerfile"})
    found = [p.name for p in iter_importable_files(root, exts)]
    assert found == ["Containerfile", "Dockerfile", "notes.md"]


def test_excluded_dirs_match_plan_anchor_a9() -> None:
    assert {
        ".venv", "node_modules", ".git", "__pycache__", ".pytest_cache", "dist", "build"
    } == EXCLUDED_DIRS


def test_added_then_unchanged_then_updated_then_pruned(db, tmp_path: Path) -> None:
    root = tmp_path / "src"
    root.mkdir()
    (root / "a.md").write_text("# A\n\nalpha\n\n## Sub\n\nmore alpha\n")
    (root / "b.md").write_text("# B\n\nbeta\n")
    llm = FakeEmbedder()
    try:
        s1 = asyncio.run(import_sources([root], llm, session=db))
        assert (s1.files, s1.added, s1.unchanged, s1.updated, s1.pruned) == (2, 2, 0, 0, 0)
        # a.md has two sections (2 chunks), b.md one (1 chunk) — content
        # chunks only; the ``is_summary`` chunks live in ``summaries``.
        assert s1.chunks == 3
        assert s1.summaries == 2  # phase 118 (A2): markdown is summarized too
        # Embeddings are stored with the configured dimension: 3 content
        # chunks + 2 ``is_summary`` chunks (one per doc, phase 118 A2).
        n = db.scalar(
            select(func.count())
            .select_from(Chunk)
            .join(Document, Document.id == Chunk.document_id)
            .where(Document.source == root.name)
        )
        assert n == 5
        for c in db.scalars(
            select(Chunk)
            .join(Document, Document.id == Chunk.document_id)
            .where(Document.source == root.name)
        ).all():
            assert c.embedding is not None and len(c.embedding) == 768

        s2 = asyncio.run(import_sources([root], llm, session=db))
        assert s2.added == 0 and s2.unchanged == 2

        (root / "a.md").write_text("# A\n\nalpha CHANGED\n")
        s3 = asyncio.run(import_sources([root], llm, session=db))
        assert s3.updated == 1 and s3.unchanged == 1
        doc = db.scalar(
            select(Document).where(Document.source == root.name, Document.path == "a.md")
        )
        assert doc is not None and "CHANGED" in doc.content

        (root / "a.md").unlink()
        s4 = asyncio.run(import_sources([root], llm, session=db, prune=True))
        assert s4.pruned == 1
        assert db.scalar(
            select(Document).where(Document.source == root.name, Document.path == "a.md")
        ) is None
        # Chunks of the pruned document are gone (FK cascade); b.md's
        # content chunk + its ``is_summary`` chunk survive (phase 118 A2).
        n_after = db.scalar(
            select(func.count())
            .select_from(Chunk)
            .join(Document, Document.id == Chunk.document_id)
            .where(Document.source == root.name)
        )
        assert n_after == 2
    finally:
        _cleanup_source(db, root.name)


def test_embedding_failure_is_logged_and_import_continues(db, tmp_path: Path) -> None:
    """A file the embedding endpoint refuses must not abort the whole KB:
    its rows are rolled back, the error is counted, and other files import."""
    root = tmp_path / "mixed"
    root.mkdir()
    (root / "bad.md").write_text("# Bad\n\npoison content that the endpoint refuses\n")
    (root / "good.md").write_text("# Good\n\nperfectly fine content\n")
    try:
        summary = asyncio.run(import_sources([root], _PoisonEmbedder(), session=db))
        assert summary.files == 2
        assert summary.errors == 1
        assert summary.added == 1  # only good.md
        # bad.md left no row and no orphan chunks behind (rolled back).
        assert db.scalar(
            select(Document).where(Document.source == root.name, Document.path == "bad.md")
        ) is None
        assert db.scalar(
            select(func.count())
            .select_from(Chunk)
            .join(Document, Document.id == Chunk.document_id)
            .where(Document.source == root.name, Document.path == "bad.md")
        ) == 0
        assert db.scalar(
            select(Document).where(Document.source == root.name, Document.path == "good.md")
        ) is not None
    finally:
        _cleanup_source(db, root.name)


def test_oversized_chunk_triggers_adaptive_rechunk(db, tmp_path: Path) -> None:
    """A URL-dense paragraph the endpoint rejects must be re-chunked smaller
    for that file only — the import still succeeds."""
    root = tmp_path / "dense"
    root.mkdir()
    # One ~1165-char paragraph: under the 1200-char hard cap, over the
    # simulated token cap. The retry at 600 chars must split it.
    para = "see https://example.com/" + "a" * 1100
    (root / "dense.md").write_text(f"# D\n\n{para}\n")
    try:
        summary = asyncio.run(import_sources([root], _CapEmbedder(), session=db))
        assert summary.errors == 0
        assert summary.added == 1
        doc = db.scalar(
            select(Document).where(Document.source == root.name, Document.path == "dense.md")
        )
        assert doc is not None
        assert len(doc.chunks) >= 2  # re-chunked smaller than the hard cap
        assert all(len(c.content) <= 1000 for c in doc.chunks)
        assert all(c.embedding is not None for c in doc.chunks)
        # The content survives the split.
        assert "".join(c.content for c in doc.chunks).count("a" * 500) >= 1
    finally:
        _cleanup_source(db, root.name)


def test_missing_source_dir_is_skipped(db, tmp_path: Path) -> None:
    llm = FakeEmbedder()
    summary = asyncio.run(import_sources([tmp_path / "missing"], llm, session=db))
    assert summary.files == 0 and summary.added == 0


def test_limit_caps_files_and_disables_prune(db, tmp_path: Path) -> None:
    root = tmp_path / "limited"
    root.mkdir()
    for name in ("a.md", "b.md", "c.md"):
        (root / name).write_text(f"# {name}\n\nbody {name}\n")
    llm = FakeEmbedder()
    try:
        summary = asyncio.run(import_sources([root], llm, limit=2, session=db, prune=True))
        assert summary.files == 2 and summary.added == 2
        # c.md was never walked, so it must NOT be pruned (prune disabled
        # under --limit) — and nothing else disappears either.
        assert summary.pruned == 0
        assert db.scalar(
            select(func.count()).select_from(Document).where(Document.source == root.name)
        ) == 2
    finally:
        _cleanup_source(db, root.name)


def test_limit_must_be_positive(db, tmp_path: Path) -> None:
    with pytest.raises(ValueError):
        asyncio.run(import_sources([tmp_path], FakeEmbedder(), limit=0, session=db))


def test_prune_is_scoped_to_the_given_sources(db, tmp_path: Path) -> None:
    src_x = tmp_path / "SourceX"
    src_y = tmp_path / "SourceY"
    src_x.mkdir()
    src_y.mkdir()
    (src_x / "x.md").write_text("# X\n\nx body\n")
    (src_y / "y.md").write_text("# Y\n\ny body\n")
    llm = FakeEmbedder()
    try:
        asyncio.run(import_sources([src_x, src_y], llm, session=db))
        # Re-import ONLY source Y (y.md removed) with prune: source X's doc
        # must survive — prune never touches sources not passed to this run.
        (src_y / "y.md").unlink()
        summary = asyncio.run(import_sources([src_y], llm, session=db, prune=True))
        assert summary.pruned == 1
        assert db.scalar(
            select(Document).where(Document.source == "SourceX", Document.path == "x.md")
        ) is not None
    finally:
        _cleanup_source(db, "SourceX")
        _cleanup_source(db, "SourceY")


def test_chunk_positions_and_titles(db, tmp_path: Path) -> None:
    root = tmp_path / "titled"
    root.mkdir()
    (root / "multi.md").write_text("# Real Title\n\n## One\n\na\n\n## Two\n\nb\n")
    (root / "noh1.md").write_text("## Only heading\n\nbody\n")
    llm = FakeEmbedder()
    try:
        asyncio.run(import_sources([root], llm, session=db))
        titles = {
            d.path: d.title
            for d in db.scalars(select(Document).where(Document.source == root.name)).all()
        }
        assert titles["multi.md"] == "Real Title"  # H1 wins
        assert titles["noh1.md"] == "noh1"  # …else the file stem
        doc = db.scalar(
            select(Document).where(Document.source == root.name, Document.path == "multi.md")
        )
        assert doc is not None
        # 0-based CONTENT positions (the ``is_summary`` chunk sits at −1,
        # phase 30/118).
        content = [c for c in doc.chunks if not c.is_summary]
        positions = sorted(c.position for c in content)
        assert positions == list(range(len(content))) and len(content) >= 2
    finally:
        _cleanup_source(db, root.name)


def test_multi_format_import_counts_per_format_and_titles_stem(db, tmp_path: Path) -> None:
    """A9 formats all import; the summary records per-format counts, and
    non-markdown titles come from the file stem (a ``#`` line is a comment
    there, not a heading)."""
    root = tmp_path / "multi"
    (root / "svc").mkdir(parents=True)
    (root / "guide.md").write_text("# Real Heading\n\nbody\n")
    (root / "svc" / "compose.yaml").write_text("# a comment\nservices:\n  gitlab: {}\n")
    (root / "svc" / "agent.py").write_text("# docstring-like comment\ndef ping():\n    return 1\n")
    (root / "inventory.json").write_text('{"hosts": []}\n')
    (root / "notes.txt").write_text("plain text notes\n")
    llm = FakeEmbedder()
    try:
        summary = asyncio.run(import_sources([root], llm, session=db))
        assert summary.files == 5
        assert summary.added == 5
        assert summary.formats == {"md": 1, "yaml": 1, "py": 1, "json": 1, "txt": 1}
        # PLAN §9 summary line: counts, highest first, ext:name pairs.
        assert summary.format_counts() == "json:1,md:1,py:1,txt:1,yaml:1"

        titles = {
            d.path: d.title
            for d in db.scalars(select(Document).where(Document.source == root.name)).all()
        }
        assert titles["guide.md"] == "Real Heading"  # markdown keeps the H1
        assert titles["svc/compose.yaml"] == "compose"  # …comment is not a heading
        assert titles["svc/agent.py"] == "agent"
        assert titles["inventory.json"] == "inventory"
        assert titles["notes.txt"] == "notes"
    finally:
        _cleanup_source(db, root.name)


def test_quadlet_and_j2_files_get_stem_titles_and_per_format_counts(
    db, tmp_path: Path
) -> None:
    """Phase 47: the ten new formats import like any other A9 format — the
    per-format counts land in the summary, every file yields content
    chunks, and a leading ``#`` line (a TOML comment in quadlet files, not
    a heading) never becomes the title: the file stem wins."""
    root = tmp_path / "quadsrc"
    (root / "quadlet").mkdir(parents=True)
    (root / "tpl").mkdir(parents=True)
    (root / "quadlet" / "web.container").write_text(
        "# RESE-QUADLET-SENTINEL-77aa\n\n[Container]\nImage=alpine\n"
    )
    (root / "quadlet" / "lan.network").write_text("# net\n[Network]\nDriver=bridge\n")
    (root / "quadlet" / "cache.volume").write_text("[Volume]\nDriver=local\n")
    (root / "quadlet" / "alpine.image").write_text("[Image]\nImages=alpine\n")
    (root / "quadlet" / "app.pod").write_text("[Pod]\nPodName=app\n")
    (root / "quadlet" / "k3s.kube").write_text("apiVersion: v1\nkind: Pod\n")
    (root / "quadlet" / "zram.swap").write_text("[Swap]\nFile=/swapfile\n")
    (root / "quadlet" / "fedora.os").write_text("[OS]\nImage=fedora\n")
    (root / "quadlet" / "edge.endpoint").write_text("[Endpoint]\nPort=8080\n")
    (root / "tpl" / "deploy.j2").write_text(
        "{% for s in services %}\n[{{ s }}]\nport = {{ s.port }}\n{% endfor %}\n"
    )
    llm = FakeEmbedder()
    try:
        summary = asyncio.run(import_sources([root], llm, session=db))
        assert summary.files == 10 and summary.added == 10
        assert summary.formats == {
            "container": 1, "network": 1, "volume": 1, "image": 1, "pod": 1,
            "kube": 1, "swap": 1, "os": 1, "endpoint": 1, "j2": 1,
        }
        docs = {
            d.path: d
            for d in db.scalars(select(Document).where(Document.source == root.name)).all()
        }
        titles = {p: d.title for p, d in docs.items()}
        assert titles == {
            # the ``#`` line is a comment in these formats — stem titles:
            "quadlet/web.container": "web",
            "quadlet/lan.network": "lan",
            "quadlet/cache.volume": "cache",
            "quadlet/alpine.image": "alpine",
            "quadlet/app.pod": "app",
            "quadlet/k3s.kube": "k3s",
            "quadlet/zram.swap": "zram",
            "quadlet/fedora.os": "fedora",
            "quadlet/edge.endpoint": "edge",
            "tpl/deploy.j2": "deploy",
        }
        for doc in docs.values():
            content = [c for c in doc.chunks if not c.is_summary]
            assert content  # every new-format file yields content chunks
            assert all(c.embedding is not None and len(c.embedding) == 768 for c in content)
    finally:
        _cleanup_source(db, root.name)


# ---------- phase 30: lite-model summaries (phase 118: every file) ----------


def test_non_markdown_file_gets_stored_and_indexed_summary(db, tmp_path: Path) -> None:
    """A ``.yaml`` file is summarized: ``documents.summary`` is set and one
    ``is_summary`` chunk (position −1, embedded) is indexed alongside the
    content chunks."""
    root = tmp_path / "sumsrc"
    root.mkdir()
    (root / "svc.yaml").write_text("alpha services:\n  gitlab:\n    port: 8929\n")
    llm = FakeEmbedder()
    try:
        summary = asyncio.run(import_sources([root], llm, session=db))
        assert summary.added == 1
        assert summary.summaries == 1
        assert summary.summary_errors == 0
        doc = db.scalar(
            select(Document).where(Document.source == root.name, Document.path == "svc.yaml")
        )
        assert doc is not None
        assert doc.summary is not None
        # Deterministic fake reply + the code-appended pointer line.
        assert doc.summary.startswith("Summary of alpha")
        assert doc.summary.endswith(f"Source: {root.name}/svc.yaml")
        schunks = [c for c in doc.chunks if c.is_summary]
        assert len(schunks) == 1
        assert schunks[0].position == -1
        assert schunks[0].content == doc.summary
        assert schunks[0].embedding is not None and len(schunks[0].embedding) == 768
        # Content chunks stay 0-based and are never flagged as summaries.
        content = [c for c in doc.chunks if not c.is_summary]
        assert sorted(c.position for c in content) == list(range(len(content)))
    finally:
        _cleanup_source(db, root.name)


def test_markdown_file_gets_stored_summary(db, tmp_path: Path) -> None:
    """Phase 118 (A2): markdown is summarized too (the phase-30 exclusion
    is retired) — ``documents.summary`` is set and one ``is_summary``
    chunk (position −1, embedded) is indexed alongside the content
    chunks."""
    root = tmp_path / "mdsrc"
    root.mkdir()
    (root / "note.md").write_text("# Note\n\nmarkdown body\n")
    llm = FakeEmbedder()
    try:
        summary = asyncio.run(import_sources([root], llm, session=db))
        assert summary.added == 1
        assert summary.summaries == 1 and summary.summary_errors == 0
        assert llm.chat_calls  # the lite model WAS asked (phase 118)
        doc = db.scalar(
            select(Document).where(Document.source == root.name, Document.path == "note.md")
        )
        assert doc is not None
        assert doc.summary is not None
        # Deterministic fake reply + the code-appended pointer line.
        assert doc.summary.startswith("Summary of")
        assert doc.summary.endswith(f"Source: {root.name}/note.md")
        schunks = [c for c in doc.chunks if c.is_summary]
        assert len(schunks) == 1
        assert schunks[0].position == -1
        assert schunks[0].content == doc.summary
        assert schunks[0].embedding is not None and len(schunks[0].embedding) == 768
        # Content chunks stay 0-based and are never flagged as summaries.
        content = [c for c in doc.chunks if not c.is_summary]
        assert sorted(c.position for c in content) == list(range(len(content)))
    finally:
        _cleanup_source(db, root.name)


# ---------- phase 118 (A2): NULL-summary backfill on the unchanged path ----------


def _clear_stored_summary(db: Session, doc: Document) -> None:
    """Simulate a NULL-summary row (a pre-phase-30 row, or a cleared
    summary): the content stays, only the summary + its chunk go away."""
    doc.summary = None
    for c in [c for c in doc.chunks if c.is_summary]:
        doc.chunks.remove(c)
    db.commit()


def test_unchanged_doc_with_null_summary_is_backfilled(db, tmp_path: Path) -> None:
    """Phase 118 (A2): an unchanged doc whose summary is NULL gets a
    summary-only backfill on the next sync: summary stored + one embedded
    ``is_summary`` chunk, counted ``summary_backfilled`` — never
    ``summaries``, never added/updated/pruned, no content re-embed."""
    root = tmp_path / "bfill"
    root.mkdir()
    (root / "note.md").write_text("# Note\n\nmarkdown body\n")
    llm = FakeEmbedder()
    try:
        first = asyncio.run(import_sources([root], llm, session=db))
        assert (first.added, first.summaries) == (1, 1)
        doc = db.scalar(
            select(Document).where(Document.source == root.name, Document.path == "note.md")
        )
        assert doc is not None and doc.summary is not None
        _clear_stored_summary(db, doc)  # the NULL-summary row the backfill targets

        embed_before = len(llm.calls)
        second = asyncio.run(import_sources([root], llm, session=db))
        assert (second.added, second.updated, second.pruned) == (0, 0, 0)
        assert second.unchanged == 1
        assert second.summary_backfilled == 1
        assert second.summaries == 0 and second.summary_errors == 0
        # No content re-embed: exactly one new embed batch, the summary
        # text only.
        assert len(llm.calls) == embed_before + 1

        db.expire_all()
        doc = db.scalar(
            select(Document).where(Document.source == root.name, Document.path == "note.md")
        )
        assert doc is not None
        assert doc.summary is not None
        assert llm.calls[-1] == [doc.summary]  # only the backfilled summary
        schunks = [c for c in doc.chunks if c.is_summary]
        assert len(schunks) == 1
        assert schunks[0].position == -1
        assert schunks[0].content == doc.summary
        assert schunks[0].embedding is not None and len(schunks[0].embedding) == 768
        # The content chunk is untouched.
        content = [c for c in doc.chunks if not c.is_summary]
        assert len(content) == 1 and content[0].embedding is not None
    finally:
        _cleanup_source(db, root.name)


def test_unchanged_doc_with_stored_summary_never_resummarizes(db, tmp_path: Path) -> None:
    """Phase 118 (A2): an unchanged doc that ALREADY has a summary (the
    third sync of the lifecycle) makes no summary LLM call at all and
    gains no chunks — owner-edited (non-NULL) summaries are never
    touched."""
    root = tmp_path / "noref"
    root.mkdir()
    (root / "note.md").write_text("# Note\n\nmarkdown body\n")
    llm = FakeEmbedder()
    try:
        asyncio.run(import_sources([root], llm, session=db))
        chat_before = len(llm.chat_calls)
        embed_before = len(llm.calls)
        chunk_before = db.scalar(
            select(func.count())
            .select_from(Chunk)
            .join(Document, Document.id == Chunk.document_id)
            .where(Document.source == root.name)
        )
        second = asyncio.run(import_sources([root], llm, session=db))
        assert second.unchanged == 1
        assert second.summary_backfilled == 0 and second.summaries == 0
        assert second.summary_errors == 0
        assert len(llm.chat_calls) == chat_before  # the model was never asked
        assert len(llm.calls) == embed_before  # no embedding of any kind
        chunk_after = db.scalar(
            select(func.count())
            .select_from(Chunk)
            .join(Document, Document.id == Chunk.document_id)
            .where(Document.source == root.name)
        )
        assert chunk_after == chunk_before  # no new chunk of any kind
    finally:
        _cleanup_source(db, root.name)


def test_unchanged_doc_with_empty_string_summary_is_never_backfilled(
    db, tmp_path: Path
) -> None:
    """Phase 118 (A2, strict ``is None``): an empty-string summary is
    owner-set (phase 57) — the backfill skips it, the ``lite`` model is
    never called, and the value stays byte-identical."""
    root = tmp_path / "emptysum"
    root.mkdir()
    (root / "note.md").write_text("# Note\n\nmarkdown body\n")
    llm = FakeEmbedder()
    try:
        asyncio.run(import_sources([root], llm, session=db))
        doc = db.scalar(
            select(Document).where(Document.source == root.name, Document.path == "note.md")
        )
        assert doc is not None
        doc.summary = ""  # the owner-set empty string (never NULL)
        db.commit()

        chat_before = len(llm.chat_calls)
        second = asyncio.run(import_sources([root], llm, session=db))
        assert second.unchanged == 1
        assert second.summary_backfilled == 0 and second.summaries == 0
        assert second.summary_errors == 0
        assert len(llm.chat_calls) == chat_before  # the model was never asked

        db.expire_all()
        doc = db.scalar(
            select(Document).where(Document.source == root.name, Document.path == "note.md")
        )
        assert doc is not None
        assert doc.summary == ""  # byte-identical — never overwritten
    finally:
        _cleanup_source(db, root.name)


def test_backfill_runs_on_manually_dated_doc_without_touching_the_date(
    db, tmp_path: Path
) -> None:
    """Phase 118 (A2, assumption 7): ``created_at_manual`` protects the
    DATE only (phase 106, D1) — a manually-dated, NULL-summary doc still
    gets its backfilled summary, and the stored date stays byte-untouched
    even though a refresh was due."""
    root = tmp_path / "manualdate"
    root.mkdir()
    (root / "note.md").write_text("# Note\n\nmarkdown body\n")
    llm = FakeEmbedder()
    try:
        asyncio.run(import_sources([root], llm, session=db))
        doc = db.scalar(
            select(Document).where(Document.source == root.name, Document.path == "note.md")
        )
        assert doc is not None
        manual = datetime(2020, 5, 4, 12, 0, 0, tzinfo=UTC)
        doc.created_at = manual
        doc.created_at_manual = True
        _clear_stored_summary(db, doc)  # the NULL-summary row the backfill targets

        second = asyncio.run(import_sources([root], llm, session=db))
        assert second.unchanged == 1
        assert second.summary_backfilled == 1 and second.summary_errors == 0
        # A date refresh WAS due (the mtime differs from the 2020
        # correction) but the manual flag withheld it — the backfill
        # never touches the date either.
        assert second.dates_updated == 0

        db.expire_all()
        doc = db.scalar(
            select(Document).where(Document.source == root.name, Document.path == "note.md")
        )
        assert doc is not None
        assert doc.summary is not None  # the backfill landed
        assert doc.created_at == manual  # byte-untouched
        assert doc.created_at_manual is True
    finally:
        _cleanup_source(db, root.name)


def test_backfill_failure_is_fail_soft_and_date_still_refreshes(
    db, tmp_path: Path
) -> None:
    """Phase 118 (A2): a backfill whose ``lite`` call fails rolls back
    its own session work only — ``summary_errors=1``, the doc row
    untouched — while the UNCHANGED path's date refresh still runs
    afterwards."""
    root = tmp_path / "bfillfail"
    root.mkdir()
    (root / "note.md").write_text("# Note\n\nmarkdown body\n")
    llm = FakeEmbedder()
    try:
        asyncio.run(import_sources([root], llm, session=db))
        doc = db.scalar(
            select(Document).where(Document.source == root.name, Document.path == "note.md")
        )
        assert doc is not None
        _clear_stored_summary(db, doc)  # the NULL-summary row the backfill targets
        # Force a date drift so the refresh is DUE on this run.
        doc.created_at = datetime(2020, 1, 1, tzinfo=UTC)
        db.commit()

        second = asyncio.run(import_sources([root], _FailingChatEmbedder(), session=db))
        assert second.unchanged == 1
        assert (second.added, second.updated, second.pruned) == (0, 0, 0)
        assert second.summary_errors == 1
        assert second.summary_backfilled == 0 and second.summaries == 0

        db.expire_all()
        doc = db.scalar(
            select(Document).where(Document.source == root.name, Document.path == "note.md")
        )
        assert doc is not None
        assert doc.summary is None  # the failed backfill left the row untouched
        assert not any(c.is_summary for c in doc.chunks)
        # …but the date refresh ran (the failure only rolled back the
        # summary's own session work).
        assert second.dates_updated == 1
        assert doc.created_at != datetime(2020, 1, 1, tzinfo=UTC)
    finally:
        _cleanup_source(db, root.name)


def test_summary_failure_is_fail_soft(db, tmp_path: Path) -> None:
    """A ``lite``-model failure must never lose the document: the file is
    fully indexed (content chunks + embeddings), ``documents.summary`` stays
    NULL, and the failure is counted in ``summary_errors``."""
    root = tmp_path / "blowup"
    root.mkdir()
    (root / "bad.txt").write_text("SUMMARY-BLOWUP the lite model chokes on this\n")
    llm = FakeEmbedder()
    try:
        summary = asyncio.run(import_sources([root], llm, session=db))
        assert summary.errors == 0  # the document itself imported fine
        assert summary.added == 1
        assert summary.summaries == 0
        assert summary.summary_errors == 1
        doc = db.scalar(
            select(Document).where(Document.source == root.name, Document.path == "bad.txt")
        )
        assert doc is not None
        assert doc.summary is None
        assert len(doc.chunks) == 1
        assert doc.chunks[0].embedding is not None  # content chunk embedded
        assert all(not c.is_summary for c in doc.chunks)
    finally:
        _cleanup_source(db, root.name)


def test_summary_chunk_is_replaced_on_reimport(db, tmp_path: Path) -> None:
    """Re-importing a changed non-markdown file keeps exactly one
    ``is_summary`` chunk — the old one is gone, the new summary is stored
    and embedded, and the content chunks stay 0-based."""
    root = tmp_path / "repl"
    root.mkdir()
    path = root / "cfg.yaml"
    path.write_text("alpha settings:\n  host: one\n")
    llm = FakeEmbedder()
    try:
        asyncio.run(import_sources([root], llm, session=db))
        path.write_text("bravo settings:\n  host: two\n")
        summary = asyncio.run(import_sources([root], llm, session=db))
        assert summary.updated == 1
        assert summary.summaries == 1 and summary.summary_errors == 0
        doc = db.scalar(
            select(Document).where(Document.source == root.name, Document.path == "cfg.yaml")
        )
        assert doc is not None
        assert doc.summary is not None and doc.summary.startswith("Summary of bravo")
        schunks = [c for c in doc.chunks if c.is_summary]
        assert len(schunks) == 1  # the old one was deleted
        assert schunks[0].position == -1
        assert schunks[0].content == doc.summary
        assert schunks[0].embedding is not None
        assert "alpha" not in schunks[0].content  # no stale summary text
        assert sorted(c.position for c in doc.chunks if not c.is_summary) == [0]
    finally:
        _cleanup_source(db, root.name)


def test_store_summary_replaces_an_existing_summary_chunk(db, tmp_path: Path) -> None:
    """Replacement unit, driven directly: with a pre-existing
    ``is_summary`` chunk in place, ``_store_summary`` deletes the old one
    and leaves exactly one (new) summary chunk + updated
    ``documents.summary`` — the at-most-one-summary invariant."""
    root = tmp_path / "direct"
    root.mkdir()
    (root / "a.yaml").write_text("alpha x\n")
    llm = FakeEmbedder()
    try:
        asyncio.run(import_sources([root], llm, session=db))
        doc = db.scalar(
            select(Document).where(Document.source == root.name, Document.path == "a.yaml")
        )
        assert doc is not None and doc.summary is not None
        assert any(c.is_summary for c in doc.chunks)  # the first import's summary
        counters = ImportSummary()
        asyncio.run(
            _store_summary(
                session=db, doc=doc, source=root.name, rel="a.yaml",
                content=doc.content, llm=llm, summary=counters,
            )
        )
        assert counters.summaries == 1 and counters.summary_errors == 0
        schunks = [c for c in doc.chunks if c.is_summary]
        assert len(schunks) == 1  # the old one was deleted
        assert schunks[0].position == -1
        assert schunks[0].content == doc.summary
        assert schunks[0].embedding is not None
    finally:
        _cleanup_source(db, root.name)


def test_store_summary_fail_soft_leaves_document_untouched(db, tmp_path: Path) -> None:
    """A ``lite`` failure inside ``_store_summary`` rolls back only the
    summary rows: the previous summary (if any) and the document survive,
    and the failure is counted."""
    root = tmp_path / "directfail"
    root.mkdir()
    (root / "a.yaml").write_text("alpha x\n")
    llm = FakeEmbedder()
    try:
        asyncio.run(import_sources([root], llm, session=db))
        doc = db.scalar(
            select(Document).where(Document.source == root.name, Document.path == "a.yaml")
        )
        assert doc is not None and doc.summary is not None
        previous_summary = doc.summary
        (root / "a.yaml").write_text("SUMMARY-BLOWUP now the lite model fails\n")
        counters = ImportSummary()
        asyncio.run(
            _store_summary(
                session=db, doc=doc, source=root.name, rel="a.yaml",
                content="SUMMARY-BLOWUP now the lite model fails\n",
                llm=llm, summary=counters,
            )
        )
        assert counters.summaries == 0 and counters.summary_errors == 1
        assert doc.summary == previous_summary  # rolled back, not nulled
        schunks = [c for c in doc.chunks if c.is_summary]
        assert len(schunks) == 1  # the old one survived the rollback
        assert schunks[0].content == previous_summary
    finally:
        _cleanup_source(db, root.name)


def test_import_summary_log_line_includes_summary_counters(
    caplog: pytest.LogCaptureFixture,
) -> None:
    """PLAN §9 summary line: the phase-30 counters sit between
    ``embed_batches`` and ``formats``; the phase-118 backfill counter
    sits between ``summary_errors`` and ``dates_updated``; the
    phase-106 date-refresh counter, the phase-122 ``images_failed``
    counter, the phase-137 ``image_context_backfilled`` counter (next
    to ``images_failed``), the phase-133 ``pdfs_screenshotted``
    counter, the phase-140 ``pdfs_pictorial`` /
    ``pdf_pictorial_pages`` counters (next to ``pdfs_screenshotted``),
    and the phase-151 ``pdf_images_deduped`` / ``pdf_images_blank_skipped``
    counters (the parser's per-PDF drop counts, next to
    ``pdf_pictorial_pages``) sit after ``dates_updated``, before
    ``formats``. Phase 187 (task 02) APPENDS a trailing ``rss_mb=``
    (LOCKED A2 — the memory the run sits at, ``?`` when unknowable); the
    pre-phase text up to and including ``formats=…`` is byte-identical.
    """
    s = ImportSummary()
    s.files, s.added, s.chunks, s.embed_batches = 3, 3, 5, 4
    s.summaries, s.summary_errors = 2, 1
    s.summary_backfilled = 1
    s.dates_updated = 0
    s.images_failed = 0
    s.svg_indexed = 4
    s.image_context_backfilled = 2
    s.pdfs_screenshotted = 1
    s.pdfs_pictorial = 2
    s.pdf_pictorial_pages = 5
    s.pdf_images_deduped = 3
    s.pdf_images_blank_skipped = 1
    s.formats = {"md": 1, "yaml": 2}
    with caplog.at_level(logging.INFO, logger="app.importer"):
        s.log()
    line = caplog.records[-1].getMessage()
    assert line.startswith(
        "import: summary files=3 added=3 updated=0 unchanged=0 pruned=0 errors=0 "
        "chunks=5 embed_batches=4 summaries=2 summary_errors=1 summary_backfilled=1 "
        "dates_updated=0 images_failed=0 svg_indexed=4 image_context_backfilled=2 "
        "pdfs_screenshotted=1 pdfs_pictorial=2 pdf_pictorial_pages=5 "
        "pdf_images_deduped=3 pdf_images_blank_skipped=1 "
        "formats=yaml:2,md:1"
    )
    # Phase 187 (task 02): the ONE appended field, `?` (unknowable) or a
    # positive integer MB — never `None`, never a negative or a float.
    assert re.fullmatch(rf".* rss_mb={MB}", line), line


def test_prune_removes_files_now_excluded_by_format_filter(db, tmp_path: Path) -> None:
    """Previously-imported junk leaves the index: a file that no longer
    matches the A9 extension filter is pruned on the next ``prune=True`` run.
    This is how dot-dir READMEs imported before the scope fix get cleaned up."""
    root = tmp_path / "cleanup"
    root.mkdir()
    (root / "keep.md").write_text("# Keep\n\nkept\n")
    (root / "junk.md.bak").write_text("old junk that was once imported\n")
    llm = FakeEmbedder()
    try:
        # Seed: import both files as if they were valid at the time.
        (root / "junk.md").write_text("old junk that was once imported\n")
        (root / "junk.md.bak").unlink()
        asyncio.run(import_sources([root], llm, session=db))
        # Rename the junk out of the A9 formats, then prune.
        (root / "junk.md").rename(root / "junk.md.bak")
        summary = asyncio.run(import_sources([root], llm, session=db, prune=True))
        assert summary.pruned == 1
        assert summary.unchanged == 1  # keep.md survived
        assert db.scalar(
            select(Document).where(Document.source == root.name, Document.path == "junk.md")
        ) is None
        assert db.scalar(
            select(Document).where(Document.source == root.name, Document.path == "keep.md")
        ) is not None
    finally:
        _cleanup_source(db, root.name)


# ---------- phase 102: extensionless name-token import ----------


def test_formats_counter_counts_extensionless_name_token(
    db, tmp_path: Path, caplog: pytest.LogCaptureFixture
) -> None:
    """Phase 102, D2 — an imported ``Dockerfile`` counts under
    ``dockerfile`` in ``summary.formats`` and the PLAN §9 line, never
    ``unknown``."""
    root = tmp_path / "extless"
    root.mkdir()
    (root / "Dockerfile").write_text("FROM alpine\n\nCMD [\"/bin/sh\"]\n")
    (root / "notes.md").write_text("# Notes\n\nbody\n")
    llm = _embedder_with_extensions("md,dockerfile")
    try:
        with caplog.at_level(logging.INFO, logger="app.importer"):
            summary = asyncio.run(import_sources([root], llm, session=db))
        assert summary.files == 2 and summary.added == 2
        assert summary.formats == {"dockerfile": 1, "md": 1}
        line = next(
            r.getMessage()
            for r in caplog.records
            if "import: summary files=" in r.getMessage()
        )
        # Phase 187 (task 02): ``formats=`` is no longer the LAST field
        # (`` rss_mb=…`` follows), but its position and value are intact.
        assert " formats=dockerfile:1,md:1" in line
        assert re.search(rf"formats=dockerfile:1,md:1 rss_mb={MB}$", line), line
        assert "unknown" not in line
    finally:
        _cleanup_source(db, root.name)


# ---------- phase 64 (task 01): optional per-file progress hook ----------


def test_progress_hook_reports_every_file_in_order_across_roots(
    db, tmp_path: Path
) -> None:
    """Multi-root, multi-file: the hook receives the exact
    ``(source, rel, done, total)`` sequence — roots in *sources* order,
    ``rel`` the same POSIX path the doc rows use, ``done`` the 1-based
    index across **all** sources, ``total`` the combined count."""
    root_a = tmp_path / "Alpha"
    root_b = tmp_path / "Beta"
    root_a.mkdir()
    (root_b / "sub").mkdir(parents=True)
    (root_a / "a1.md").write_text("# A1\n\na one\n")
    (root_a / "a2.md").write_text("# A2\n\na two\n")
    (root_a / "a1.md").write_text("# A1\n\na one\n")
    (root_b / "sub" / "b1.md").write_text("# B1\n\nb one\n")
    events: list[tuple[str, str, int, int]] = []

    def progress(source: str, rel: str, done: int, total: int) -> None:
        events.append((source, rel, done, total))

    try:
        summary = asyncio.run(
            import_sources([root_a, root_b], FakeEmbedder(), session=db, progress=progress)
        )
        assert summary.files == 3 and summary.added == 3
        assert events == [
            ("Alpha", "a1.md", 1, 3),
            ("Alpha", "a2.md", 2, 3),
            ("Beta", "sub/b1.md", 3, 3),  # POSIX rel, sorted within the root
        ]
    finally:
        _cleanup_source(db, "Alpha")
        _cleanup_source(db, "Beta")


def test_no_progress_means_no_prewalk(
    db, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    """``progress=None`` callers pay no extra pass: the real walker is hit
    exactly once per source root (one pass — as before phase 64), proven
    with a counting sentinel; with the hook it is hit twice (the phase-194
    pre-scan for the pending set — the phase-64 stats pre-walk when
    ``BOR_SYNC_PENDING_COUNT=0`` — plus the processing pass).

    Phase 194 (task 01) moved the hook's pair, and this fixture is the
    second run over an ALREADY-INDEXED file: nothing is pending, so the
    pair is ``(0, 0)`` — the denominator is the predicted change set and an
    unchanged KB predicts nothing (the UI then renders no count part). The
    walk COUNT is what this test owns and it is untouched: one pass without
    the hook, two with it."""
    root = tmp_path / "nowalk"
    root.mkdir()
    (root / "a.md").write_text("# A\n\na\n")
    real_walker = importer.iter_importable_files
    walk_calls = 0

    def counting(
        r: Path,
        extensions: frozenset[str],
        excluded: frozenset[str] = EXCLUDED_DIRS,
        ignore: tuple[str, ...] = (),
        include_hidden: bool = False,
        image_extensions: frozenset[str] = frozenset(),
    ) -> list[Path]:
        # Phase 89: the walker gained the ``ignore`` keyword; phase 105:
        # the ``include_hidden`` flag; phase 122: the ``image_extensions``
        # set — the sentinel accepts (and forwards) all three to stay a
        # drop-in.
        nonlocal walk_calls
        walk_calls += 1
        return real_walker(
            r, extensions, excluded, ignore, include_hidden, image_extensions
        )

    monkeypatch.setattr(importer, "iter_importable_files", counting)
    try:
        summary = asyncio.run(import_sources([root], FakeEmbedder(), session=db))
        assert summary.files == 1
        assert walk_calls == 1  # exactly one pass — the pre-change behaviour
        walk_calls = 0
        events: list[tuple[str, str, int, int]] = []
        summary2 = asyncio.run(
            import_sources(
                [root], FakeEmbedder(), session=db,
                progress=lambda s, r, d, t: events.append((s, r, d, t)),
            )
        )
        assert summary2.files == 1
        assert walk_calls == 2  # the count pass (pending set / pre-walk) + walk
        # phase 194: a.md is unchanged, so the predicted change set is empty
        # — ``total`` 0 and the position never advances. With the kill switch
        # at ``0`` this same run reports the phase-193 walk position (1 of 1);
        # pinned in tests/unit/test_sync_pending_count.py.
        assert events == [(root.name, "a.md", 0, 0)]
    finally:
        _cleanup_source(db, root.name)


def test_progress_hook_counts_unchanged_and_error_files(db, tmp_path: Path) -> None:
    """The hook fires *before* ``_index_file``: a file whose embedding
    fails (and one that re-imports as unchanged) is still reported as the
    current file — the sequence covers every importable file.

    Phase 194 (task 01) retargeted the pair to the PENDING position (the
    owner's "1216 remaining" fix), which reads exactly right here. The
    first run is fresh: both files are predicted, ``1 of 2`` then ``2 of 2``
    — identical to the pre-phase numbers (A1's first-import parity). The
    second run is the incremental case: ``bad.md`` rolled back with its
    embed error so its row is still missing (predicted again, the ONLY
    work), ``good.md`` is unchanged so it holds the position instead of
    advancing it — ``1 of 1``, ``1 of 1``: every file is still reported as
    the current file, and the count is the work."""
    root = tmp_path / "progress-mixed"
    root.mkdir()
    (root / "bad.md").write_text("# Bad\n\npoison content the endpoint refuses\n")
    (root / "good.md").write_text("# Good\n\nperfectly fine content\n")
    expected = [(root.name, "bad.md", 1, 2), (root.name, "good.md", 2, 2)]
    try:
        events: list[tuple[str, str, int, int]] = []
        first = asyncio.run(
            import_sources(
                [root], _PoisonEmbedder(), session=db,
                progress=lambda s, r, d, t: events.append((s, r, d, t)),
            )
        )
        # bad.md was already reported (done=1) when its embed raised —
        # no file silently disappears from the sequence.
        assert events == expected
        assert first.errors == 1 and first.added == 1
        # Re-run: good.md is now unchanged, bad.md is retried and fails
        # again — both still report as the current file, but only bad.md is
        # pending, so the denominator is 1 and good.md holds the position.
        events.clear()
        second = asyncio.run(
            import_sources(
                [root], _PoisonEmbedder(), session=db,
                progress=lambda s, r, d, t: events.append((s, r, d, t)),
            )
        )
        assert events == [(root.name, "bad.md", 1, 1), (root.name, "good.md", 1, 1)]
        assert second.errors == 1 and second.unchanged == 1
    finally:
        _cleanup_source(db, root.name)


def test_progress_hook_with_limit_keeps_full_total(db, tmp_path: Path) -> None:
    """The debug ``limit`` path is unchanged for the hook: it fires only
    for processed files (``done`` never exceeds the limit), while
    ``total`` stays the FULL pre-walk count — an incomplete walk must not
    misreport the denominator."""
    root = tmp_path / "progress-limited"
    root.mkdir()
    for name in ("a.md", "b.md", "c.md"):
        (root / name).write_text(f"# {name}\n\nbody {name}\n")
    events: list[tuple[str, str, int, int]] = []
    try:
        summary = asyncio.run(
            import_sources(
                [root], FakeEmbedder(), limit=2, session=db,
                progress=lambda s, r, d, t: events.append((s, r, d, t)),
            )
        )
        assert summary.files == 2
        assert events == [(root.name, "a.md", 1, 3), (root.name, "b.md", 2, 3)]
    finally:
        _cleanup_source(db, root.name)


# ---------- phase 127 (task 01): ImportSummary.changed_paths ----------


class _VisionLLM(FakeEmbedder):
    """The phase-122 mock VISION client (the test_pdf pattern): the
    multimodal describe call (``user`` content = a part list) gets a fixed
    description; text (``lite``) calls keep the ``FakeEmbedder``
    behaviour."""

    DESCRIPTION = "A test image: a solid swatch of color."

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


def _png_bytes(width: int, height: int, rgb: tuple[float, float, float]) -> bytes:
    """One solid-color PNG built with PyMuPDF (the phase-126 no-binary-
    fixtures convention)."""
    pix = fitz.Pixmap(fitz.csRGB, fitz.IRect(0, 0, width, height), 0)
    for y in range(height):
        for x in range(width):
            pix.set_pixel(x, y, rgb)
    return pix.tobytes("png")


def _build_one_page_pdf(path: Path, text: str) -> None:
    """One temp single-page PDF (the phase-126 convention — no binary
    fixtures)."""
    doc = fitz.open()
    doc.new_page().insert_text((72, 72), text)
    doc.save(path)
    doc.close()


def test_import_summary_changed_paths_defaults_empty() -> None:
    """Completion criterion: default construction gives an empty change
    set — no existing construction site migrates."""
    assert ImportSummary().changed_paths == []


def test_changed_paths_records_added_updated_and_pruned_in_order(
    db, tmp_path: Path
) -> None:
    """The contract (phase 127, task 01): one ``(source, path)`` entry per
    ADDED, UPDATED, or PRUNED document — added/updated in walk order, the
    pruned pair last (``_prune`` runs after the walk) — and
    ``len(changed_paths) == added + updated + pruned``."""
    root = tmp_path / "changed"
    root.mkdir()
    for name in ("a.md", "b.md", "c.md"):
        (root / name).write_text(f"# {name}\n\nv1\n")
    llm = FakeEmbedder()
    try:
        source = root.name
        first = asyncio.run(import_sources([root], llm, session=db))
        assert first.added == 3 and first.pruned == 0
        assert first.changed_paths == [
            (source, "a.md"),
            (source, "b.md"),
            (source, "c.md"),
        ]
        # Run 2: update a.md, add d.md + e.md, delete c.md (→ pruned).
        (root / "a.md").write_text("# a\n\nv2 changed\n")
        (root / "d.md").write_text("# d\n\nnew d\n")
        (root / "e.md").write_text("# e\n\nnew e\n")
        (root / "c.md").unlink()
        second = asyncio.run(import_sources([root], llm, session=db, prune=True))
        assert (second.added, second.updated, second.unchanged, second.pruned) == (
            2, 1, 1, 1,
        )
        assert second.changed_paths == [
            (source, "a.md"),  # updated — first in walk order
            (source, "d.md"),  # added
            (source, "e.md"),  # added
            (source, "c.md"),  # pruned — the _prune pass runs after the walk
        ]
        assert len(second.changed_paths) == (
            second.added + second.updated + second.pruned
        )
    finally:
        _cleanup_source(db, root.name)


def test_changed_paths_empty_on_unchanged_rerun(db, tmp_path: Path) -> None:
    """An all-unchanged re-run (prune included) records nothing."""
    root = tmp_path / "nochange"
    root.mkdir()
    (root / "a.md").write_text("# a\n\nstable\n")
    (root / "b.md").write_text("# b\n\nstable\n")
    llm = FakeEmbedder()
    try:
        asyncio.run(import_sources([root], llm, session=db))
        second = asyncio.run(import_sources([root], llm, session=db, prune=True))
        assert second.unchanged == 2
        assert second.changed_paths == []
    finally:
        _cleanup_source(db, root.name)


def test_changed_paths_empty_on_backfill_rerun(db, tmp_path: Path) -> None:
    """A backfill (phase 118: unchanged doc, NULL summary) is NOT a content
    change: the add still records, the backfill re-run stays empty."""
    root = tmp_path / "backfill"
    root.mkdir()
    (root / "a.yaml").write_text("alpha x\n")
    try:
        first = asyncio.run(import_sources([root], _FailingChatEmbedder(), session=db))
        assert first.added == 1 and first.summary_errors == 1
        assert first.changed_paths == [(root.name, "a.yaml")]  # the ADD records
        doc = db.scalar(
            select(Document).where(Document.source == root.name, Document.path == "a.yaml")
        )
        assert doc is not None and doc.summary is None
        second = asyncio.run(import_sources([root], FakeEmbedder(), session=db))
        assert second.unchanged == 1 and second.summary_backfilled == 1
        assert second.changed_paths == []
    finally:
        _cleanup_source(db, root.name)


def test_changed_paths_empty_on_date_only_rerun(db, tmp_path: Path) -> None:
    """A date-only refresh (phase 106, D4: content untouched, mtime moved)
    is NOT a content change: the re-run records nothing."""
    root = tmp_path / "dateonly"
    root.mkdir()
    path = root / "a.md"
    path.write_text("# a\n\nstable\n")
    os.utime(path, (0, datetime(2020, 1, 1, tzinfo=UTC).timestamp()))
    llm = FakeEmbedder()
    try:
        first = asyncio.run(import_sources([root], llm, session=db))
        assert first.added == 1 and first.dates_updated == 0
        assert first.changed_paths == [(root.name, "a.md")]
        # The content is untouched — only the mtime moved (the D4
        # date-refresh on the unchanged path).
        os.utime(path, (0, datetime(2021, 6, 15, tzinfo=UTC).timestamp()))
        second = asyncio.run(import_sources([root], llm, session=db))
        assert second.unchanged == 1 and second.dates_updated == 1
        assert second.changed_paths == []
    finally:
        _cleanup_source(db, root.name)


def test_changed_paths_records_a_changed_image(db, tmp_path: Path) -> None:
    """The image increment site (phase 122 path): a NEW image and a
    byte-CHANGED image each record their ``(source, rel)``."""
    root = tmp_path / "imgsrc"
    root.mkdir()
    (root / "pic.png").write_bytes(_png_bytes(1, 1, (1.0, 0.0, 0.0)))
    llm = _VisionLLM()
    llm.settings = Settings(
        _env_file=None,  # pyright: ignore[reportCallIssue]
        images=True,
        image_dir=str(tmp_path / "images"),
    )
    try:
        source = root.name
        first = asyncio.run(import_sources([root], llm, session=db))
        assert first.added == 1 and first.images_failed == 0
        assert first.changed_paths == [(source, "pic.png")]
        # The BYTES changed (a new color) — the image digest is over the
        # bytes, so this is an UPDATE.
        (root / "pic.png").write_bytes(_png_bytes(1, 1, (0.0, 1.0, 1.0)))
        second = asyncio.run(import_sources([root], llm, session=db))
        assert second.updated == 1
        assert second.changed_paths == [(source, "pic.png")]
    finally:
        _cleanup_source(db, root.name)


def test_changed_paths_records_a_changed_pdf(db, tmp_path: Path) -> None:
    """The PDF increment site (phase 126 path): a NEW PDF and a text-
    CHANGED PDF each record their ``(source, rel)``."""
    root = tmp_path / "pdfsrc"
    root.mkdir()
    pdf_path = root / "doc.pdf"
    _build_one_page_pdf(pdf_path, "alpha page one")
    llm = _VisionLLM()
    llm.settings = Settings(
        _env_file=None,  # pyright: ignore[reportCallIssue]
        images=True,
        image_dir=str(tmp_path / "images"),
        import_extensions="md,txt,pdf",
    )
    try:
        source = root.name
        first = asyncio.run(import_sources([root], llm, session=db))
        assert first.added == 1
        assert first.changed_paths == [(source, "doc.pdf")]
        # The extracted TEXT changed — the PDF digest is over it, so this
        # is an UPDATE.
        _build_one_page_pdf(pdf_path, "bravo page one, revised")
        second = asyncio.run(import_sources([root], llm, session=db))
        assert second.updated == 1
        assert second.changed_paths == [(source, "doc.pdf")]
    finally:
        _cleanup_source(db, root.name)


def test_prune_records_changed_paths_only_when_summary_given(db, tmp_path: Path) -> None:
    """The ``_prune`` recording seam: with NO summary (the pre-phase
    count-only contract) nothing is recorded; with one, each ACTUALLY
    deleted doc appends its ``(source, path)`` (skipped docs never do)."""
    root = tmp_path / "prunit"
    root.mkdir()
    (root / "gone.md").write_text("# gone\n\nbye\n")
    (root / "gone2.md").write_text("# gone2\n\nbye\n")
    llm = FakeEmbedder()
    try:
        source = root.name
        asyncio.run(import_sources([root], llm, session=db))
        (root / "gone.md").unlink()
        (root / "gone2.md").unlink()
        # Default call (summary=None): count only, no recording.
        pruned = _prune(db, {source}, {(source, "gone2.md")})
        assert pruned == 1
        assert db.scalar(
            select(Document).where(Document.source == source, Document.path == "gone.md")
        ) is None
        # With a summary: the deleted doc records its pair.
        counters = ImportSummary()
        pruned = _prune(db, {source}, set(), summary=counters)
        assert pruned == 1
        assert counters.changed_paths == [(source, "gone2.md")]
    finally:
        _cleanup_source(db, root.name)


# ---------- phase 183 (task 02): the prune deletes IN the database ----------


def test_document_chunks_relationship_is_passive_deletes() -> None:
    """``Document.chunks`` is ``passive_deletes=True`` (phase 183, LOCKED A1).

    The mapper-level pin is the cheapest place to catch the keyword being
    dropped: without it ``session.delete(doc)`` SELECTs every chunk of the
    document before deleting it, hydrating every 768-dim
    ``Chunk.embedding`` (~24.8 KiB of Python each) — the shape that made a
    3,008-document prune cost ~720 MB and locked up the 2026-10-15 VM
    (3,904 MB, no swap). The other half of the contract — the FK really is
    ``ON DELETE CASCADE`` — is pinned live by
    ``tests/integration/test_prune_cascade_statements.py``
    (``test_chunks_fk_cascades_at_database_level``), so these two tests
    fail together the moment either side drifts.
    """
    rel = Document.__mapper__.relationships["chunks"]
    assert rel.passive_deletes is True
    # The ORPHAN semantics ``_replace_content_chunks``' re-embed relies on
    # (``doc.chunks = [...]`` drops the old rows) stay intact (A2).
    assert "delete-orphan" in str(rel.cascade)


# ---------- phase 129 (task 02): on_chunk_progress plumbing ----------


class _BatchingEmbedder(FakeEmbedder):
    """Mimics ``LLMClient``'s top-level token-budgeted batching (the
    task-01 contract on the importer side): *texts* is split into
    consecutive groups of *batch_size* and the hook fires
    ``(first_chunk_1based, total)`` before EACH group — the group is then
    embedded as one request (``super().embed`` with ``on_progress=None``:
    this class owns the hook sequence, so the base's single-batch
    announce is suppressed per group)."""

    def __init__(self, batch_size: int) -> None:
        super().__init__()
        self.batch_size = batch_size

    async def embed(
        self,
        texts: list[str],
        *,
        on_progress: Callable[[int, int], None] | None = None,
    ) -> list[list[float]]:
        out: list[list[float]] = []
        for start in range(0, len(texts), self.batch_size):
            group = texts[start : start + self.batch_size]
            if on_progress is not None:
                on_progress(start + 1, len(texts))
            out.extend(await super().embed(group, on_progress=None))
        return out


def _markdown_with_sections(stem: str, n: int) -> str:
    """One markdown file with a preamble (h1 title + intro) and *n* short
    ``##`` sections — each section is under the 2000-char target, so the
    file chunks to exactly *n* + 1 content chunks (the preamble + one per
    section; verified against ``chunk_document`` below)."""
    lines = [f"# {stem}", "", f"Intro line for {stem.lower()}.", ""]
    for i in range(1, n + 1):
        lines += [f"## {stem} {i}", "", f"Body text for section {i}.", ""]
    return "\n".join(lines)


def _chunk_count(content: str, rel: str) -> int:
    """The content-chunk count the importer's pipeline will produce for
    *content* (the same ``chunk_document`` call and defaults as
    ``_replace_content_chunks``: ``max(400, chunk_target_chars)`` target,
    ``chunk_overlap_chars`` overlap)."""
    s = Settings(_env_file=None)  # pyright: ignore[reportCallIssue]
    return len(chunk_document(content, rel, max(400, s.chunk_target_chars), s.chunk_overlap_chars))


def test_on_chunk_progress_reports_per_file_batch_sequences(
    db, tmp_path: Path
) -> None:
    """A multi-batch file's embed position is fully observable: the hook
    fires before each embed batch with ``(first_chunk_1based, total)`` —
    ``total`` is the file's content-chunk count, the first indices climb
    by the previous batch's size, and a SECOND file in the same run gets
    its OWN sequence (the two files' totals never mix)."""
    root = tmp_path / "chunkprog"
    root.mkdir()
    content_a = _markdown_with_sections("Alpha", 8)  # preamble + 8 = 9 chunks
    content_b = _markdown_with_sections("Beta", 4)  # preamble + 4 = 5 chunks
    (root / "alpha.md").write_text(content_a)
    (root / "beta.md").write_text(content_b)
    n_a, n_b = _chunk_count(content_a, "alpha.md"), _chunk_count(content_b, "beta.md")
    assert (n_a, n_b) == (9, 5)

    batch_size = 3

    def expected(n: int) -> list[tuple[int, int]]:
        return [(start + 1, n) for start in range(0, n, batch_size)]

    # The task's multi-batch case: file A spans >= 2 batches.
    assert len(expected(n_a)) >= 2

    llm = _BatchingEmbedder(batch_size=batch_size)
    chunk_events: list[tuple[int, int]] = []
    file_events: list[tuple[str, str, int, int]] = []
    try:
        summary = asyncio.run(
            import_sources(
                [root], llm, session=db,
                progress=lambda s, r, d, t: file_events.append((s, r, d, t)),
                on_chunk_progress=lambda f, t: chunk_events.append((f, t)),
            )
        )
        assert (summary.added, summary.chunks) == (2, n_a + n_b)
    finally:
        _cleanup_source(db, root.name)

    # The per-file ``progress`` hook names each file in walk order ...
    assert file_events == [
        (root.name, "alpha.md", 1, 2),
        (root.name, "beta.md", 2, 2),
    ]
    # ... and the chunk sequence is exactly file A's batches followed by
    # file B's OWN restarted sequence (its totals never leak into A's).
    assert chunk_events == expected(n_a) + expected(n_b)


def test_on_chunk_progress_single_batch_file_and_summary_excluded(
    db, tmp_path: Path
) -> None:
    """A file whose chunks fit in ONE embed batch reports exactly one
    ``(1, N)`` event, and the document's summary-chunk embed (the SECOND
    ``llm.embed`` call) NEVER fires the hook — the counter counts content
    chunks only (A6: no ``1/1`` noise after the real progress ends)."""
    root = tmp_path / "chunkprog-one"
    root.mkdir()
    content = _markdown_with_sections("Gamma", 3)  # preamble + 3 = 4 chunks
    (root / "gamma.md").write_text(content)
    n = _chunk_count(content, "gamma.md")
    assert n == 4
    events: list[tuple[int, int]] = []
    llm = FakeEmbedder()
    try:
        summary = asyncio.run(
            import_sources(
                [root], llm, session=db,
                on_chunk_progress=lambda f, t: events.append((f, t)),
            )
        )
        assert summary.added == 1
    finally:
        _cleanup_source(db, root.name)
    assert events == [(1, n)]
    # Two embed calls happened (the content batch + the excluded summary
    # chunk) — the hook fired for the first only.
    assert [len(c) for c in llm.calls] == [n, 1]


def test_on_chunk_progress_image_branch_reports_one_of_one(
    db, tmp_path: Path
) -> None:
    """The image branch's ONE description-chunk embed reports exactly
    ``(1, 1)`` — it IS the file's only chunk — and the mirrored
    summary-chunk embed never fires the hook. The REAL description seam
    is used (the ``_VisionLLM`` multimodal ``chat``)."""
    root = tmp_path / "imgchunk"
    root.mkdir()
    (root / "pic.png").write_bytes(_png_bytes(4, 4, (0.8, 0.2, 0.2)))
    events: list[tuple[int, int]] = []
    llm = _VisionLLM()
    llm.settings = Settings(
        _env_file=None,  # pyright: ignore[reportCallIssue]
        images=True,
        image_dir=str(tmp_path / "images"),
    )
    try:
        summary = asyncio.run(
            import_sources(
                [root], llm, session=db,
                on_chunk_progress=lambda f, t: events.append((f, t)),
            )
        )
        assert (summary.files, summary.added) == (1, 1)
    finally:
        _cleanup_source(db, root.name)
    assert events == [(1, 1)]
    # The content embed (the 1 description chunk) + the excluded
    # summary-chunk embed — the hook fired for the first only.
    assert [len(c) for c in llm.calls] == [1, 1]


def test_on_chunk_progress_pdf_branch_reports_content_chunks(
    db, tmp_path: Path
) -> None:
    """The PDF branch's CONTENT embed is the primary target: a two-page
    PDF (two ``Page N:`` chunks) reports ``(1, 2)`` once through the
    hook, and the summary-chunk embed never fires it."""
    root = tmp_path / "pdfchunk"
    root.mkdir()
    doc = fitz.open()
    doc.new_page().insert_text((72, 72), "First page text about networking.")
    doc.new_page().insert_text((72, 72), "Second page text about storage.")
    doc.save(root / "report.pdf")
    doc.close()
    events: list[tuple[int, int]] = []
    llm = _VisionLLM()
    llm.settings = Settings(
        _env_file=None,  # pyright: ignore[reportCallIssue]
        images=True,
        image_dir=str(tmp_path / "images"),
        import_extensions="md,txt,pdf",
        # Phase 140: the pre-phase-140 shape pin — this fixture's pages
        # are sparse (the default-ON pictorial pass would append
        # illustration blocks and add vision calls); the pass is
        # covered in tests/unit/test_pictorial_import.py.
        pdf_pictorial=False,
    )
    try:
        summary = asyncio.run(
            import_sources(
                [root], llm, session=db,
                on_chunk_progress=lambda f, t: events.append((f, t)),
            )
        )
        assert (summary.files, summary.added) == (1, 1)
    finally:
        _cleanup_source(db, root.name)
    assert events == [(1, 2)]
    # The content embed (the 2 page chunks, one batch) + the excluded
    # summary-chunk embed — the hook fired for the first only.
    assert [len(c) for c in llm.calls] == [2, 1]


# ---------- phase 133, task 03: the page-screenshot fallback ----------


class _PageVisionLLM(FakeEmbedder):
    """The phase-133 mock VISION client — a STABLE vision model: the
    MULTIMODAL describe call (``user`` content = a part list — the
    shape ``describe_image`` sends) gets a canned reply DETERMINED BY
    THE IMAGE'S BYTES (a short sha256 of the data URL's base64 — the
    same rendered page always gets the same reply, ACROSS RUNS; the
    pages' bytes differ, so the replies differ and the page order
    stays checkable, and the digest-stability test's "stable vision
    model" premise holds). Text (``lite``) calls keep the
    ``FakeEmbedder`` behaviour. Every vision call's model is recorded
    (``vision_models``)."""

    def __init__(self) -> None:
        super().__init__()
        self.vision_calls = 0
        self.vision_models: list[str | None] = []

    @staticmethod
    def reply_for(image_bytes: bytes) -> str:
        """The canned reply a vision call for *image_bytes* gets —
        deterministic in the bytes (tests build the expected content
        with the same helper)."""
        return f"Page {hashlib.sha256(image_bytes).hexdigest()[:8]} described."

    async def chat(self, messages, model=None):  # pyright: ignore[reportIncompatibleMethodOverride]
        self.chat_calls.append(list(messages))
        user = next((m["content"] for m in messages if m.get("role") == "user"), "")
        if isinstance(user, list):
            self.vision_calls += 1
            self.vision_models.append(model)
            for part in user:
                if part.get("type") == "image_url":
                    b64 = part["image_url"]["url"].rsplit(",", 1)[1]
                    return self.reply_for(base64.b64decode(b64))
            raise LLMError("no image part in the multimodal message (test bug)")
        first = user.split()
        return "Summary of " + (first[0] if first else "<empty>")


class _FirstPageFailsVisionLLM(FakeEmbedder):
    """A vision client that REJECTS the first multimodal call (a dead
    page's vision request) and then describes — drives the per-page
    fail-soft (the dead page becomes the placeholder, the next page
    still gets its description)."""

    def __init__(self) -> None:
        super().__init__()
        self.vision_calls = 0

    async def chat(self, messages, model=None):  # pyright: ignore[reportIncompatibleMethodOverride]
        self.chat_calls.append(list(messages))
        user = next((m["content"] for m in messages if m.get("role") == "user"), "")
        if isinstance(user, list):
            self.vision_calls += 1
            if self.vision_calls == 1:
                raise LLMError("simulated dead vision call (test sentinel)")
            return "Page two described fine."
        first = user.split()
        return "Summary of " + (first[0] if first else "<empty>")


class _AlwaysFailsVisionLLM(FakeEmbedder):
    """A vision client that REJECTS every multimodal call (the dead
    vision endpoint — the per-page fail-soft's total-failure case:
    every page becomes the placeholder and the doc lands empty)."""

    def __init__(self) -> None:
        super().__init__()
        self.vision_calls = 0

    async def chat(self, messages, model=None):  # pyright: ignore[reportIncompatibleMethodOverride]
        self.chat_calls.append(list(messages))
        user = next((m["content"] for m in messages if m.get("role") == "user"), "")
        if isinstance(user, list):
            self.vision_calls += 1
            raise LLMError("simulated dead vision endpoint (test sentinel)")
        first = user.split()
        return "Summary of " + (first[0] if first else "<empty>")


def _pdf_settings(tmp_path: Path, images: bool = False) -> Settings:
    """Settings admitting the ``pdf`` import token (the code DEFAULT set
    has no ``pdf`` — the phase-133 task-01 trap: a default fake would
    silently skip the PDF) with the image knobs at their test values
    (``images`` off unless pinned otherwise — the fallback's own vision
    calls need no image copy, only the phase-122 embedded-image pass
    does). Phase 140: ``pdf_pictorial`` is pinned OFF — these suites
    pin the PRE-phase-140 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; ``tests/unit/test_pictorial_import.py`` covers the pass
    itself, with the kill switch in both states)."""
    return Settings(
        _env_file=None,  # pyright: ignore[reportCallIssue] (kwarg exists at runtime)
        import_extensions="md,txt,pdf",
        images=images,
        image_dir=str(tmp_path / "images"),
        pdf_pictorial=False,
    )


def _build_garbage_pdf(path: Path, pages: int = 2) -> None:
    """A REAL tiny garbage PDF: every page carries ONLY XML-ish lines
    (``<foo:pN>i</foo:pN>`` — page-specific tags, so the rendered page
    PNGs differ and the stable fake's replies do too) — the gate's XML
    signal fails deterministically (100% of the text lines look like
    tags; the ``---PAGE-N---`` separators are not tag-shaped). The text
    layer is real EXTRACTABLE text — this is the FINDING's class (the
    pre-126 raw-XMP soup), not the empty-layer class."""
    doc = fitz.open()
    for p in range(1, pages + 1):
        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(path)
    doc.close()


def _pdf_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 test_garbage_pdf_takes_the_page_screenshot_fallback(
    db, tmp_path: Path, caplog: pytest.LogCaptureFixture
) -> None:
    """The owner's specified path end to end: a PDF whose text layer is
    XML garbage fails the gate, every page is rendered to a PNG and
    read by the vision model, and the doc lands with the
    ``---PAGE-N---``-anchored page descriptions (PAGE ORDER pinned by
    the differing canned replies) — the page-aware chunker works
    unchanged on the separators, the digest is over the FINAL content,
    and the counter + the loud warning (the FINDING's signature) fire.
    ``is_image`` stays False (it is a PDF document whose content
    happens to be vision text)."""
    root = tmp_path / "garbage"
    root.mkdir()
    _build_garbage_pdf(root / "garbage.pdf")
    llm = _PageVisionLLM()
    llm.settings = _pdf_settings(tmp_path)
    # The stable fake's expected replies (rendered page bytes → reply;
    # page order pinned: page i's bytes carry page i's description).
    pngs = importer.render_pdf_pages(root / "garbage.pdf")
    d1, d2 = _PageVisionLLM.reply_for(pngs[0]), _PageVisionLLM.reply_for(pngs[1])
    try:
        with caplog.at_level(logging.INFO, logger="app.importer"):
            summary = asyncio.run(import_sources([root], llm, session=db))
        assert (summary.files, summary.added, summary.errors) == (1, 1, 0)
        assert summary.pdfs_screenshotted == 1
        assert llm.vision_calls == 2  # one vision call per page

        doc = _pdf_doc(db, "garbage", "garbage.pdf")
        assert doc.is_image is False and doc.image_path is None
        assert doc.content == f"---PAGE-1---\n{d1}\n\n---PAGE-2---\n{d2}"
        assert d1 != d2, "the pages' bytes differ — so do their replies"
        # The digest runs over the FINAL (fallback) content — the
        # phase-126 digest rule, unchanged.
        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{d1}",
            f"Page 2:\n{d2}",
        ]
        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

        # The loud warning (PLAN §9, the FINDING's signature) — the
        # required line, exact.
        warnings = [
            r for r in caplog.records if "text layer unusable" in r.getMessage()
        ]
        assert len(warnings) == 1 and warnings[0].levelno == logging.WARNING
        assert warnings[0].getMessage() == (
            "pdf: text layer unusable for garbage/garbage.pdf — "
            "falling back to 2 page screenshots"
        )
        # The vision calls went to the CHAT model (the phase-122 seam).
        assert llm.vision_models == [llm.settings.llm_chat_model] * 2
    finally:
        _cleanup_source(db, "garbage")


def test_clean_pdf_never_burns_a_fallback_vision_call(
    db, tmp_path: Path, caplog: pytest.LogCaptureFixture
) -> None:
    """The gate is the ONLY gate: a clean text layer never reaches the
    fallback — ``pdfs_screenshotted`` stays 0 and the multimodal chat
    is NEVER called (the images toggle is off, so the only vision seam
    in play is the fallback itself; the phase-126 happy path is
    byte-identical — the extracted text is the content, no fallback
    warning, no ``Page N —`` placeholders)."""
    root = tmp_path / "clean"
    root.mkdir()
    _build_one_page_pdf(root / "clean.pdf", "Clean prose about the homelab.")
    llm = _PageVisionLLM()
    llm.settings = _pdf_settings(tmp_path)
    try:
        with caplog.at_level(logging.INFO, logger="app.importer"):
            summary = asyncio.run(import_sources([root], llm, session=db))
        assert (summary.files, summary.added, summary.errors) == (1, 1, 0)
        assert summary.pdfs_screenshotted == 0
        assert llm.vision_calls == 0
        doc = _pdf_doc(db, "clean", "clean.pdf")
        assert doc.content == "---PAGE-1---\nClean prose about the homelab."
        assert not [
            r for r in caplog.records if "text layer unusable" in r.getMessage()
        ]
    finally:
        _cleanup_source(db, "clean")


def test_fallback_is_fail_soft_per_page(
    db, tmp_path: Path, caplog: pytest.LogCaptureFixture
) -> None:
    """A dead vision call on ONE page → the deterministic placeholder
    line (``page N — description unavailable``) for that page, the
    other page still gets its description — the doc still lands (a
    dead vision endpoint must not sink the doc, the phase-122
    fail-soft spirit). The vision seam logged its failure (the
    phase-122 warning); the fallback counted once."""
    root = tmp_path / "partfail"
    root.mkdir()
    _build_garbage_pdf(root / "partfail.pdf", pages=2)
    llm = _FirstPageFailsVisionLLM()
    llm.settings = _pdf_settings(tmp_path)
    try:
        with caplog.at_level(logging.INFO, logger="app.summarizer"):
            summary = asyncio.run(import_sources([root], llm, session=db))
        assert (summary.files, summary.added, summary.errors) == (1, 1, 0)
        assert summary.pdfs_screenshotted == 1
        assert llm.vision_calls == 2  # both pages were attempted
        doc = _pdf_doc(db, "partfail", "partfail.pdf")
        assert doc.content == (
            "---PAGE-1---\npage 1 — description unavailable\n\n"
            "---PAGE-2---\nPage two described fine."
        )
        # The phase-122 seam's failure warning (the dead page's signal).
        assert [
            r for r in caplog.records if "image description failed" in r.getMessage()
        ]
    finally:
        _cleanup_source(db, "partfail")


def test_fallback_all_pages_failed_lands_empty_content(
    db, tmp_path: Path, caplog: pytest.LogCaptureFixture
) -> None:
    """A garbage PDF whose vision calls ALL fail (the dead vision
    endpoint) → the existing EMPTY-CONTENT path: the doc lands with
    ``""`` (the empty byte string's digest, no content chunks), the
    fallback was still taken (``pdfs_screenshotted == 1`` — the gate
    fired), and the phase-126 "no extractable text" warning fires —
    the doc really is empty. A dead endpoint must not sink the sync."""
    root = tmp_path / "allfail"
    root.mkdir()
    _build_garbage_pdf(root / "allfail.pdf", pages=1)
    llm = _AlwaysFailsVisionLLM()
    llm.settings = _pdf_settings(tmp_path)
    try:
        with caplog.at_level(logging.INFO, logger="app.importer"):
            summary = asyncio.run(import_sources([root], llm, session=db))
        assert (summary.files, summary.added, summary.errors) == (1, 1, 0)
        assert summary.pdfs_screenshotted == 1
        assert llm.vision_calls == 1
        doc = _pdf_doc(db, "allfail", "allfail.pdf")
        assert doc.content == ""
        assert doc.content_hash == hashlib.sha256(b"").hexdigest()
        chunks = db.scalars(
            select(Chunk).where(Chunk.document_id == doc.id).order_by(Chunk.position)
        ).all()
        assert [c for c in chunks if not c.is_summary] == []
        warnings = [
            r for r in caplog.records if "no extractable text" in r.getMessage()
        ]
        assert len(warnings) == 1 and warnings[0].levelno == logging.WARNING
    finally:
        _cleanup_source(db, "allfail")


def test_fallback_unrenderable_page_gets_the_placeholder(
    db, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    """A page that cannot be RENDERED (``b""`` at its index — the
    :func:`render_pdf_pages` fail-soft slot) gets the deterministic
    placeholder while the good pages keep their descriptions — the
    page index never shifts (``---PAGE-N---`` stays truthful), and the
    unrenderable page burns no vision call."""
    root = tmp_path / "halfblind"
    root.mkdir()
    _build_garbage_pdf(root / "halfblind.pdf", pages=3)
    monkeypatch.setattr(
        importer,
        "render_pdf_pages",
        lambda _path, dpi=150: [b"\x89PNG-page-1", b"", b"\x89PNG-page-3"],
    )
    llm = _PageVisionLLM()
    llm.settings = _pdf_settings(tmp_path)
    d1 = _PageVisionLLM.reply_for(b"\x89PNG-page-1")
    d3 = _PageVisionLLM.reply_for(b"\x89PNG-page-3")
    try:
        summary = asyncio.run(import_sources([root], llm, session=db))
        assert summary.pdfs_screenshotted == 1
        assert llm.vision_calls == 2  # the blank page burned no vision call
        doc = _pdf_doc(db, "halfblind", "halfblind.pdf")
        assert doc.content == (
            f"---PAGE-1---\n{d1}\n\n"
            "---PAGE-2---\npage 2 — description unavailable\n\n"
            f"---PAGE-3---\n{d3}"
        )
    finally:
        _cleanup_source(db, "halfblind")


def test_fallback_rerun_unchanged_fast_path_no_reburn(
    db, tmp_path: Path, caplog: pytest.LogCaptureFixture
) -> None:
    """Phase 147 (task 02) — the owner's core complaint, fixed: a
    re-sync of an unchanged garbage PDF (same raw bytes) takes the
    FAST PATH at the top of ``_index_file`` (the file hash matches)
    — the fallback does NOT re-run: the gate never fires
    (``pdfs_screenshotted`` stays 0 on run 2), no re-extract, no
    re-render, and the vision calls are NOT re-issued (the previous
    "vision calls re-issued before the unchanged check can skip them"
    no-cache cost is gone). The row re-imports as UNCHANGED — no
    re-chunk, no re-embed, no KB change (not ``updated``, no changed
    path) — and the doc's content/digest are byte-stable."""
    root = tmp_path / "stable"
    root.mkdir()
    _build_garbage_pdf(root / "stable.pdf")
    llm = _PageVisionLLM()
    llm.settings = _pdf_settings(tmp_path)
    try:
        s1 = asyncio.run(import_sources([root], llm, session=db))
        assert (s1.added, s1.pdfs_screenshotted) == (1, 1)
        assert llm.vision_calls == 2  # one per page — the first index
        embeds1 = len(llm.calls)
        doc1 = _pdf_doc(db, "stable", "stable.pdf")
        content1, digest1 = doc1.content, doc1.content_hash

        s2 = asyncio.run(import_sources([root], llm, session=db))
        # The fast path fired (file hash match) — the fallback never
        # ran on run 2: no gate, no render, no vision re-issued, no
        # re-embed, no KB change.
        assert (s2.added, s2.updated, s2.unchanged) == (0, 0, 1)
        assert s2.pdfs_screenshotted == 0  # the gate never re-fired
        assert llm.vision_calls == 2  # NO re-issued vision call
        assert len(llm.calls) == embeds1, "unchanged: no re-embed"
        assert s2.changed_paths == [], "unchanged: no KB change recorded"

        doc2 = _pdf_doc(db, "stable", "stable.pdf")
        assert doc2.content == content1  # byte-stable content
        assert doc2.content_hash == digest1  # byte-stable digest
    finally:
        _cleanup_source(db, "stable")


def test_fallback_embedded_images_still_join_the_pipeline(
    db, tmp_path: Path
) -> None:
    """Both, no conflict (the phase design): a fallback PDF with the
    images toggle ON ALSO runs the phase-126 embedded-image pass — the
    screenshots are the CONTENT (the fallback's vision calls: one per
    page), the extracted images are the citation figures (the
    phase-122 pass's vision calls: one per image). The content keeps
    the page descriptions; the image doc is separate."""
    root = tmp_path / "both"
    root.mkdir()
    # Garbage text (the gate fires) + an embedded image on page 2
    # (the phase-126 A3 pass indexes it).
    doc = fitz.open()
    page1 = doc.new_page()
    for i in range(10):
        page1.insert_text((72, 72 + i * 14), f"<foo:bar>{i}</foo:bar>")
    page2 = doc.new_page()
    for i in range(10):
        page2.insert_text((72, 72 + i * 14), f"<foo:baz>{i}</foo:baz>")
    pix = fitz.Pixmap(fitz.csRGB, fitz.IRect(0, 0, 40, 30), 0)
    pix.clear_with(200)
    page2.insert_image(fitz.Rect(72, 220, 112, 250), stream=pix.tobytes("png"))
    doc.save(root / "both.pdf")
    doc.close()
    llm = _PageVisionLLM()
    llm.settings = _pdf_settings(tmp_path, images=True)
    try:
        summary = asyncio.run(import_sources([root], llm, session=db))
        # The PDF (screenshot content) + its one embedded image doc.
        assert (summary.files, summary.added, summary.errors) == (1, 2, 0)
        assert summary.pdfs_screenshotted == 1
        assert summary.images_failed == 0
        # 2 fallback page descriptions + 1 embedded-image description.
        assert llm.vision_calls == 3

        pdf_doc = _pdf_doc(db, "both", "both.pdf")
        pngs = importer.render_pdf_pages(root / "both.pdf")
        d1 = _PageVisionLLM.reply_for(pngs[0])
        d2 = _PageVisionLLM.reply_for(pngs[1])
        assert d1 != d2
        assert pdf_doc.content == f"---PAGE-1---\n{d1}\n\n---PAGE-2---\n{d2}"
        # The embedded image (the clear_with(200) 40x30 PNG) gets its
        # OWN stable reply — the third vision call, the citation figure.
        image_doc = _pdf_doc(db, "both", "both.pdf#page=2#img=0.png")
        assert image_doc.is_image is True
        assert image_doc.content != d1 and image_doc.content != d2
    finally:
        _cleanup_source(db, "both")


# ---------- phase 152 (task 02): on_image_progress plumbing ----------

#: A text block comfortably over the phase-140 100-char sparseness bar
#: (a USABLE layer, zero pictorial candidates even with the toggle on
#: — the "no vision span" class). Inserted wrapped at 60 chars: a
#: single long line would be CLIPPED at the page edge on extraction,
#: while the wrapped multi-line read round-trips (the phase-140
#: fixture's class property).
_P152_DENSE_TEXT = (
    "The quadlet unit file defines the container interface, the image, "
    "the network, and the restart policy. Every key maps to a docker "
    "flag in the generated service file, and comments use the usual "
    "semicolon form throughout the whole document."
)
assert len(_P152_DENSE_TEXT.strip()) > 100  # the class property, not a typo


def _p152_solid_png(value: int) -> bytes:
    """A 40x30 solid-color PNG (channel *value*): every value used
    below stays under the phase-151 blank gate's 235 mean floor (kept,
    never blank-skipped), and distinct values are distinct images (the
    dedupe keeps every one)."""
    pix = fitz.Pixmap(fitz.csRGB, fitz.IRect(0, 0, 40, 30), 0)
    pix.clear_with(value)
    return pix.tobytes("png")


def _build_image_pdf(
    path: Path,
    values: Sequence[int],
    dense: bool = True,
    repeat_first: bool = False,
) -> None:
    """A REAL one-page PDF: *dense* a text block over the 100-char bar
    (a USABLE layer — the fallback never fires; zero pictorial
    candidates even with the toggle on) carrying *values* the
    solid-color embedded images (one placement each, distinct).
    ``dense=False`` inserts the sparse ``BESTA`` marker instead (a
    pictorial candidate under the default 100-char bar).
    ``repeat_first`` adds a SECOND placement of the first image (the
    same stream — one xref, the parser's repeated-placement class,
    phase 151: the dedupe drops it, the position count is
    post-dedupe)."""
    doc = fitz.open()
    page = doc.new_page()
    text = _P152_DENSE_TEXT if dense else "BESTA"
    for i, line in enumerate(textwrap.wrap(text, 60)):
        page.insert_text((72, 72 + i * 14), line)
    y = 300
    for value in values:
        page.insert_image(fitz.Rect(72, y, 112, y + 30), stream=_p152_solid_png(value))
        y += 40
    if repeat_first:
        page.insert_image(
            fitz.Rect(150, 300, 190, 330), stream=_p152_solid_png(values[0])
        )
    doc.save(path)
    doc.close()


def _build_pictorial_image_pdf(path: Path, values: Sequence[int]) -> None:
    """A REAL two-page PDF — the ONE continuous counter's two-span
    case: page 1 is SPARSE (``BESTA`` — a pictorial candidate under
    the default 100-char bar), page 2 is dense (never a candidate) and
    carries *values* the solid-color embedded images. Both the
    phase-140 pictorial pass (before the document commit) and the
    phase-126 embedded-image pass (after it) fire on it."""
    doc = fitz.open()
    doc.new_page().insert_text((72, 72), "BESTA")
    page = doc.new_page()
    for i, line in enumerate(textwrap.wrap(_P152_DENSE_TEXT, 60)):
        page.insert_text((72, 72 + i * 14), line)
    y = 300
    for value in values:
        page.insert_image(fitz.Rect(72, y, 112, y + 30), stream=_p152_solid_png(value))
        y += 40
    doc.save(path)
    doc.close()


def _build_garbage_image_pdf(path: Path, values: Sequence[int]) -> None:
    """A REAL two-page garbage-text PDF (every page XML-ish — the
    phase-133 gate fires, the page-screenshot fallback runs) with
    *values* the solid-color embedded images on page 2 — the fallback
    span + the embedded span, ONE continuous counter."""
    doc = fitz.open()
    for p in range(1, 3):
        page = doc.new_page()
        for i in range(10):
            page.insert_text((72, 72 + i * 14), f"<foo:p{p}>{i}</foo:p{p}>")
    y = 400
    for value in values:
        doc[1].insert_image(
            fitz.Rect(72, y, 112, y + 30), stream=_p152_solid_png(value)
        )
        y += 40
    doc.save(path)
    doc.close()


def _p152_summary_fields(s: ImportSummary) -> tuple:
    """One run's observable outcome (the byte-identity pin's left
    side): every count a hook could plausibly move + the changed
    paths' RELS (the source names are equal by construction — the two
    roots carry the same directory name)."""
    return (
        s.files, s.added, s.updated, s.unchanged, s.pruned, s.errors,
        s.chunks, s.embed_batches, s.summaries, s.summary_errors,
        s.images_failed, s.pdfs_screenshotted, s.pdfs_pictorial,
        s.pdf_pictorial_pages, s.pdf_images_deduped, s.pdf_images_blank_skipped,
        tuple(rel for _, rel in s.changed_paths),
    )


def _p152_rows(db, source: str) -> list[tuple]:
    """A source's document rows (the byte-identity pin's right side),
    identity-blind: the fields a hook could plausibly move."""
    docs = db.scalars(select(Document).where(Document.source == source)).all()
    return sorted(
        (d.path, d.title, d.content, d.content_hash, d.is_image, d.summary,
         d.image_context)
        for d in docs
    )



def test_on_image_progress_embedded_pass_climbs_one_per_image(
    db, tmp_path: Path
) -> None:
    """The phase-152 core pin: a PDF with N DISTINCT embedded images
    (one placement each — post-dedupe, post-blank-gate: the work
    actually about to happen) fires the hook ONCE per image, 1-based,
    with the span's total, at the START of each image's pipeline step:
    ``(1, N) … (N, N)``. A repeated placement of the first image (the
    parser's dedupe, phase 151) adds NO position (the total is the
    post-dedupe count), a clean usable text layer burns no fallback or
    pictorial span (the embedded pass's totals stand alone), and the
    phase-147 fast path on a re-sync (raw bytes unchanged) burns no
    vision call and fires the hook never."""
    root = tmp_path / "imgprog"
    root.mkdir()
    values = (90, 130, 170, 210)  # 4 distinct images, all under the blank floor
    _build_image_pdf(root / "report.pdf", values, repeat_first=True)
    n = len(values)
    llm = _PageVisionLLM()
    llm.settings = _pdf_settings(tmp_path, images=True)
    events: list[tuple[int, int]] = []
    try:
        summary = asyncio.run(
            import_sources(
                [root], llm, session=db,
                on_image_progress=lambda pos, total: events.append((pos, total)),
            )
        )
        # The PDF + its N image docs; the repeat placement deduped.
        assert (summary.files, summary.added) == (1, 1 + n)
        assert summary.pdf_images_deduped == 1
        assert (summary.images_failed, summary.pdfs_screenshotted,
                summary.pdfs_pictorial) == (0, 0, 0)
        # The embedded pass is the ONLY vision span: N slow calls.
        assert llm.vision_calls == n
        # The core pin: once per image, 1-based, the span's total.
        assert events == [(i, n) for i in range(1, n + 1)]
        # The phase-147 fast path: a re-sync (bytes unchanged) burns no
        # vision call and fires the hook never — the events list grows
        # not at all.
        summary2 = asyncio.run(
            import_sources(
                [root], llm, session=db,
                on_image_progress=lambda pos, total: events.append((pos, total)),
            )
        )
        assert summary2.unchanged == 1
        assert llm.vision_calls == n
        assert len(events) == n
    finally:
        _cleanup_source(db, root.name)


def test_on_image_progress_pictorial_then_embedded_is_one_counter(
    db, tmp_path: Path
) -> None:
    """ASSUMPTION A3: the spans of ONE file are ONE continuous counter
    — the embedded pass starts where the pictorial pass ended (``base
    = images_done_so_far``): the positions climb ``1…C+N`` with NO
    restart and never above the combined total (the pictorial span's
    total is its candidate-page count, the embedded span's total the
    combined count)."""
    root = tmp_path / "imgprog-two"
    root.mkdir()
    values = (90, 130, 170)
    _build_pictorial_image_pdf(root / "ikea.pdf", values)
    # The span inputs, computed IN the test (no hardcoded lists — the
    # phase-140 idiom): the deterministic gate's candidates and the
    # post-dedupe/post-gate image count.
    pdf = root / "ikea.pdf"
    page_texts = extract_pdf_page_texts(pdf)
    n_cand = len(pictorial_pages(page_texts, min_chars=100))
    images, _ = extract_pdf_images_with_stats(pdf)
    n_img = len(images)
    assert (n_cand, n_img) == (1, len(values))
    llm = _PageVisionLLM()
    llm.settings = Settings(
        _env_file=None,  # pyright: ignore[reportCallIssue]
        import_extensions="md,txt,pdf",
        images=True,
        image_dir=str(tmp_path / "images"),
        pdf_pictorial=True,  # default on — the pass runs on the sparse page
    )
    events: list[tuple[int, int]] = []
    try:
        summary = asyncio.run(
            import_sources(
                [root], llm, session=db,
                on_image_progress=lambda pos, total: events.append((pos, total)),
            )
        )
        assert (summary.files, summary.added) == (1, 1 + n_img)
        assert (summary.pdfs_pictorial, summary.pdf_pictorial_pages) == (1, 1)
        # The continuous counter: 1…C (total C) then C+1…C+N (total
        # C+N) — no restart, never above the combined total.
        assert events == (
            [(i, n_cand) for i in range(1, n_cand + 1)]
            + [(n_cand + i, n_cand + n_img) for i in range(1, n_img + 1)]
        )
    finally:
        _cleanup_source(db, root.name)


def test_on_image_progress_fallback_span_climbs_into_embedded(
    db, tmp_path: Path
) -> None:
    """The owner's stuck-label class: a garbage-text PDF's per-page
    screenshot describes (the phase-133 fallback span — the span that
    froze the label for minutes) report the rendered-page count
    (``1…M`` with total ``M``) and hand their reported count to the
    embedded pass, which continues the SAME counter (``M+1`` with
    total ``M+N``) — every slow call has a position before it."""
    root = tmp_path / "imgprog-fb"
    root.mkdir()
    _build_garbage_image_pdf(root / "garbage.pdf", (90,))
    llm = _PageVisionLLM()
    llm.settings = _pdf_settings(tmp_path, images=True)
    events: list[tuple[int, int]] = []
    try:
        summary = asyncio.run(
            import_sources(
                [root], llm, session=db,
                on_image_progress=lambda pos, total: events.append((pos, total)),
            )
        )
        assert summary.pdfs_screenshotted == 1
        assert (summary.files, summary.added, summary.images_failed) == (1, 2, 0)
        # 2 fallback pages (total 2) + 1 embedded image continuing at
        # 3 (total 3) — ONE counter across the two spans.
        assert events == [(1, 2), (2, 2), (3, 3)]
        # Every slow call (2 pages + 1 image) was preceded by a
        # position — the label never sits frozen on a busy run.
        assert llm.vision_calls == 3
    finally:
        _cleanup_source(db, root.name)


def test_on_image_progress_text_only_pdf_never_fires(
    db, tmp_path: Path
) -> None:
    """A clean-text PDF with NO embedded images fires the hook NEVER,
    with BOTH default-on toggles: the garbage gate stays closed (no
    fallback span), the pictorial pass's deterministic gate finds zero
    candidates (a no-op — the clean-text byte-identity class), and the
    embedded pass's list is empty. The skipped spans report nothing
    and advance the counter not at all (there is no vision call at
    all)."""
    root = tmp_path / "imgprog-none"
    root.mkdir()
    _build_image_pdf(root / "clean.pdf", ())  # dense text, zero images
    llm = _PageVisionLLM()
    llm.settings = Settings(
        _env_file=None,  # pyright: ignore[reportCallIssue]
        import_extensions="md,txt,pdf",
        images=True,
        image_dir=str(tmp_path / "images"),
        pdf_pictorial=True,  # default on — the pass runs and no-ops
    )
    events: list[tuple[int, int]] = []
    try:
        summary = asyncio.run(
            import_sources(
                [root], llm, session=db,
                on_image_progress=lambda pos, total: events.append((pos, total)),
            )
        )
        assert (summary.files, summary.added) == (1, 1)
        assert (summary.pdfs_screenshotted, summary.pdfs_pictorial) == (0, 0)
        assert llm.vision_calls == 0  # no span ran — zero slow calls
        assert events == []
    finally:
        _cleanup_source(db, root.name)


def test_on_image_progress_toggle_off_reports_nothing(
    db, tmp_path: Path
) -> None:
    """The ``images`` toggle OFF skips the embedded span entirely (the
    prune guard keeps any existing image docs) — a PDF that WOULD
    carry N embedded images reports nothing (a skipped span advances
    the counter not at all) and indexes as text only."""
    root = tmp_path / "imgprog-off"
    root.mkdir()
    _build_image_pdf(root / "off.pdf", (90, 130))
    llm = _PageVisionLLM()
    llm.settings = _pdf_settings(tmp_path, images=False)
    events: list[tuple[int, int]] = []
    try:
        summary = asyncio.run(
            import_sources(
                [root], llm, session=db,
                on_image_progress=lambda pos, total: events.append((pos, total)),
            )
        )
        assert (summary.files, summary.added) == (1, 1)
        assert llm.vision_calls == 0
        assert events == []
    finally:
        _cleanup_source(db, root.name)


def test_on_image_progress_none_run_is_byte_identical(
    db, tmp_path: Path
) -> None:
    """The regression pin: a hook-less run (the CLI/chat default —
    ``on_image_progress=None``, every call site a guarded no-op) lands
    byte-identical to a run WITH the hook: the same documents (path,
    title, content, digest, image flag, summary, context) and the same
    summary counters. The hook observes; it never changes the run."""
    # Both roots carry the SAME directory name — the source (the
    # walk's root.name) is the document's provenance and part of the
    # stored image context, so the comparison needs it equal.
    root_a = tmp_path / "a" / "same_src"
    root_a.mkdir(parents=True)
    root_b = tmp_path / "b" / "same_src"
    root_b.mkdir(parents=True)
    values = (90, 130, 170, 210)
    _build_image_pdf(root_a / "same.pdf", values)
    _build_image_pdf(root_b / "same.pdf", values)
    llm_a, llm_b = _PageVisionLLM(), _PageVisionLLM()
    llm_a.settings = _pdf_settings(tmp_path, images=True)
    llm_b.settings = _pdf_settings(tmp_path, images=True)
    events: list[tuple[int, int]] = []
    try:
        # Run A (hook-less) first, snapshot its rows, clear the shared
        # source name, then run B (hooked) — the rows line up 1:1.
        summary_a = asyncio.run(import_sources([root_a], llm_a, session=db))
        rows_a = _p152_rows(db, "same_src")
        _cleanup_source(db, "same_src")
        summary_b = asyncio.run(
            import_sources(
                [root_b], llm_b, session=db,
                on_image_progress=lambda pos, total: events.append((pos, total)),
            )
        )
        rows_b = _p152_rows(db, "same_src")
        # The hook ran in run B (N image positions, the span's total)
        # — and changed nothing observable.
        assert events == [(i, len(values)) for i in range(1, len(values) + 1)]
        assert _p152_summary_fields(summary_a) == _p152_summary_fields(summary_b)
        assert rows_a == rows_b
    finally:
        _cleanup_source(db, "same_src")


class _InterleavingVisionLLM(_PageVisionLLM):
    """A ``_PageVisionLLM`` variant that records each vision call in
    a SHARED event sequence (phase 152, task 04): the importer fires
    the ``on_image_progress`` hook BEFORE the
    ``_index_one_embedded_image`` describe seam, so the sequence reads
    hook, vision, hook, vision, … — the interleaving pin (the label
    moves at the START of each slow call, never after it)."""

    def __init__(self, events: list[tuple]) -> None:
        super().__init__()
        self.events = events

    async def chat(self, messages, model=None):  # pyright: ignore[reportIncompatibleMethodOverride]
        reply = await super().chat(messages, model)
        user = next((m["content"] for m in messages if m.get("role") == "user"), "")
        if isinstance(user, list):
            self.events.append(("vision", self.vision_calls))
        return reply


def test_on_image_progress_hook_fires_before_each_describe_call(
    db, tmp_path: Path
) -> None:
    """Phase 152 (task 04): the hook fires BEFORE the corresponding
    describe call (the contract: at the START of each slow call, never
    after it) — the interleaving pinned by recording events in the
    fake: for a PDF with N embedded images the sequence reads
    hook(1,N), vision(1), hook(2,N), vision(2), …, hook(N,N),
    vision(N) (the fake's vision call IS the describe seam the hook
    precedes — the label moves while the describe is about to start,
    so a busy run never reads frozen)."""
    root = tmp_path / "imgprog-order"
    root.mkdir()
    values = (90, 130, 170)
    _build_image_pdf(root / "order.pdf", values)
    n = len(values)
    events: list[tuple] = []
    llm = _InterleavingVisionLLM(events)
    llm.settings = _pdf_settings(tmp_path, images=True)
    try:
        summary = asyncio.run(
            import_sources(
                [root], llm, session=db,
                on_image_progress=lambda pos, total: events.append(("image", pos, total)),
            )
        )
        assert (summary.files, summary.added) == (1, 1 + n)
        assert llm.vision_calls == n  # one describe per image
        # The interleaving: hook BEFORE each describe call, in order.
        expected: list[tuple] = []
        for i in range(1, n + 1):
            expected.append(("image", i, n))
            expected.append(("vision", i))
        assert events == expected
    finally:
        _cleanup_source(db, root.name)


def test_on_image_progress_blank_gate_excludes_skipped_images_from_the_total(
    db, tmp_path: Path
) -> None:
    """Phase 152 (task 04, the phase-151 interaction): images skipped
    by the dedupe/blank gate are NOT counted in the hook's total —
    the total is the work actually about to happen (the pass's
    post-dedupe, post-gate list). A PDF with 2 real + 1 near-white
    image (250 — the blank gate's class, mean ≥ 235): the gate
    (default ON) drops the near-white one BEFORE the pass — the hook
    fires for the 2 kept with total 2 (the skipped image burns no
    vision call, no position); with the kill switch OFF
    (``pdf_image_skip_blank=False`` — the pre-phase-151 behavior) the
    near-white image is indexed too — the hook fires for all 3 with
    total 3 (the total is the work about to happen under the knob's
    state)."""
    values_a = (90, 130, 250)   # 250 = near-white (the blank gate's class)
    values_b = (110, 150, 250)  # distinct bytes — a different document set
    # Both roots carry the SAME directory name — the source is part of
    # the stored image context, so the runs line up 1:1 (the house
    # cleanup idiom separates them).
    root_a = tmp_path / "a" / "same_src"
    root_a.mkdir(parents=True)
    root_b = tmp_path / "b" / "same_src"
    root_b.mkdir(parents=True)
    _build_image_pdf(root_a / "blank.pdf", values_a)
    _build_image_pdf(root_b / "blank.pdf", values_b)
    llm_a = _PageVisionLLM()
    llm_a.settings = _pdf_settings(tmp_path, images=True)  # skip_blank default ON
    events_a: list[tuple[int, int]] = []
    try:
        summary_a = asyncio.run(
            import_sources(
                [root_a], llm_a, session=db,
                on_image_progress=lambda pos, total: events_a.append((pos, total)),
            )
        )
        # The blank gate dropped the near-white image BEFORE the
        # pass — the total is the 2 actually about to happen.
        assert summary_a.pdf_images_blank_skipped == 1
        assert (summary_a.files, summary_a.added) == (1, 3)
        assert llm_a.vision_calls == 2
        assert events_a == [(1, 2), (2, 2)]
        # The knob OFF: the near-white image is indexed too — the
        # total is all 3 (the pre-phase behavior, the work about to
        # happen under the knob's state). Fresh source (the house
        # cleanup idiom — run B's rows would otherwise be updates of
        # run A's rows, same synthetic image paths).
        _cleanup_source(db, "same_src")
        llm_b = _PageVisionLLM()
        llm_b.settings = Settings(
            _env_file=None,  # pyright: ignore[reportCallIssue]
            import_extensions="md,txt,pdf",
            images=True,
            image_dir=str(tmp_path / "images"),
            pdf_pictorial=False,
            pdf_image_skip_blank=False,
        )
        events_b: list[tuple[int, int]] = []
        summary_b = asyncio.run(
            import_sources(
                [root_b], llm_b, session=db,
                on_image_progress=lambda pos, total: events_b.append((pos, total)),
            )
        )
        assert summary_b.pdf_images_blank_skipped == 0
        assert (summary_b.files, summary_b.added) == (1, 4)
        assert llm_b.vision_calls == 3
        assert events_b == [(1, 3), (2, 3), (3, 3)]
    finally:
        _cleanup_source(db, "same_src")