"""Phase 147, task 02 — the unchanged fast path (unit matrix).

The owner's TODO L4: "Local sources, like PDFs, are synced every time
even if they haven't changed. This results in really long sync times
for PDFs that are mostly pictoral." The fix: the sha256 of the RAW
source file's bytes (:func:`app.rag.importer._file_hash` — streamed in
1 MiB chunks) is computed ONCE at the top of ``_index_file``, BEFORE
the image/text/PDF dispatch, and compared against
``documents.file_hash`` (migration 0025). A match returns through the
shared unchanged path (``_unchanged_doc``) with ZERO llm calls (embed
+ chat, both counted through the fakes), ZERO PDF extraction, and
ZERO vision work — while the preserved unchanged semantics (the
phase-118 summary-NULL backfill, the phase-106 D1 manual-date lock,
the D4 date refresh, the counts, the log wordings) survive verbatim.

Pinned here (the load-bearing pins are the ZERO-llm-call assertions on
the second sync):

* the fast path fires for all three document classes (text, image,
  garbage-layer PDF, pictorial PDF) and returns before any branch code;
* a changed file (any byte) re-indexes fully and re-stamps the hash;
* a bytes-changed/content-same file (a PDF metadata edit) re-runs the
  branch's pipeline but lands unchanged (no content change) and
  stamps the new hash — the next sync is fast;
* a pre-phase row (``file_hash`` NULL) runs the branch's content-digest
  path (today's behavior) and gets STAMPED — the next sync is fast;
* the D1 manual-date lock and the D4 date refresh (may go older,
  counts ``dates_updated``) hold on the fast path;
* the phase-118 backfill on the fast path runs from the STORED
  ``doc.content`` (one ``is_summary`` chunk, no content re-embed);
* the streamed hash is transparent (a >1 MiB file hashes identically
  to ``hashlib.sha256(bytes).hexdigest()``).

Phase 163 (task 03) extends this file with the PDF strip RE-GATE the
unchanged paths now run (``_regate_pdf_strip``) — the pre-phase "the
copies persist" assumption is what broke in production (a row claiming
``page_count = 36`` over ZERO files, every page 404ing forever): the
re-gate re-derives the strip from the current bytes + settings (ZERO
vision/LLM calls), heals missing copies, re-stamps a changed class,
clears a strip that no longer qualifies (kill switch included), stays
fail-soft, logs only when it ACTS, and costs a non-strip file zero PDF
parsing (spy-pinned).

The E2E surface (the idempotent second sync through the real UI +
mock LLM) is pinned in ``tests/e2e/test_sync_changed_only.py``
(task 03).
"""
from __future__ import annotations

import asyncio
import hashlib
import logging
import os
from datetime import UTC, datetime, timedelta
from pathlib import Path
from typing import Any

import pymupdf as fitz
import pytest
from sqlalchemy import select

from app.config import Settings
from app.models import Document
from app.rag import importer
from app.rag.importer import _file_hash, import_sources
from tests.fakes import FakeEmbedder
from tests.unit.test_importer import (
    _build_garbage_pdf,
    _PageVisionLLM,
    _pdf_settings,
    _png_bytes,
    _VisionLLM,
)
from tests.unit.test_pictorial_import import (
    _build_dense_pdf,
    _build_ikea_like_pdf,
    _PictorialLLM,
    _settings,
)

#: mtime granularity tolerance (the ``test_importer_dates.py``
#: convention).
_TOL = timedelta(milliseconds=50)

OLD_2021 = datetime(2021, 6, 1, 12, 0, 0, tzinfo=UTC)


def _utime(path: Path, when: datetime) -> None:
    ts = when.timestamp()
    os.utime(path, (ts, ts))


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


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


# ---------- the streamed hash -------------------------------------------------


def test_file_hash_streamed_matches_direct_hash(tmp_path: Path) -> None:
    """The chunking is transparent: a >1 MiB file (1.5 MiB here —
    spans multiple 1 MiB reads) hashes identically to
    ``hashlib.sha256(bytes).hexdigest()``, an empty file hashes the
    empty byte string, and the result is a 64-char lowercase hex
    digest (the ``documents.file_hash`` column width)."""
    block = hashlib.sha256(b"seed-147").digest()  # 32 B
    data = block * 49152  # 1.5 MiB — well past one 1 MiB chunk
    big = tmp_path / "big.bin"
    big.write_bytes(data)
    expected = hashlib.sha256(data).hexdigest()
    assert len(expected) == 64 and expected == expected.lower()
    assert _file_hash(big) == expected

    empty = tmp_path / "empty.bin"
    empty.write_bytes(b"")
    assert _file_hash(empty) == hashlib.sha256(b"").hexdigest()


# ---------- the fast path: zero llm on the second sync ------------------------


def test_text_unchanged_second_sync_costs_zero_llm(db, tmp_path: Path) -> None:
    """Text class: index a file, capture the llm call count (embed +
    chat both counted); re-run → ``unchanged == 1``, ZERO new llm
    calls, ``files == 1``, the doc's ``indexed_at``/``content_hash``
    untouched, and ``file_hash`` stamped after run 1 (equal to
    ``_file_hash`` of the raw bytes)."""
    root = tmp_path / "fasttext"
    root.mkdir()
    path = root / "note.md"
    path.write_text("# Note\n\nmarkdown body\n")
    llm = FakeEmbedder()
    try:
        first = asyncio.run(import_sources([root], llm, session=db))
        assert (first.added, first.unchanged, first.files) == (1, 0, 1)
        doc = _doc(db, root.name, "note.md")
        # Stamped after run 1 — the raw-bytes digest.
        assert doc.file_hash == _file_hash(path)
        embeds1, chats1 = len(llm.calls), len(llm.chat_calls)
        indexed1, digest1 = doc.indexed_at, doc.content_hash

        second = asyncio.run(import_sources([root], llm, session=db))
        assert (second.added, second.updated, second.unchanged, second.files) == (0, 0, 1, 1)
        # THE pin: zero new llm calls of any kind.
        assert len(llm.calls) == embeds1, "fast path: no re-embed"
        assert len(llm.chat_calls) == chats1, "fast path: no lite/vision call"
        assert second.changed_paths == []  # no KB change recorded

        db.expire_all()
        doc2 = _doc(db, root.name, "note.md")
        assert doc2.indexed_at == indexed1  # the re-index did not happen
        assert doc2.content_hash == digest1
        assert doc2.file_hash == _file_hash(path)
    finally:
        _cleanup_source(db, root.name)


def test_garbage_pdf_unchanged_second_sync_costs_zero_vision(db, tmp_path: Path) -> None:
    """Garbage-layer PDF class (the owner's core complaint): a no-text
    layer's PDF indexed through the page-screenshot fallback (two
    vision calls); re-run → ``unchanged == 1``, ZERO new llm calls —
    the fallback vision did NOT re-fire (the pre-phase "vision
    re-issued before the unchanged check can skip them" cost is
    gone), the gate never re-fires, and the content row is
    byte-identical."""
    root = tmp_path / "fastgarbage"
    root.mkdir()
    path = root / "garbage.pdf"
    _build_garbage_pdf(path)
    llm = _PageVisionLLM()
    llm.settings = _pdf_settings(tmp_path)
    try:
        first = asyncio.run(import_sources([root], llm, session=db))
        assert (first.added, first.pdfs_screenshotted) == (1, 1)
        assert llm.vision_calls == 2  # one per page — the first index
        doc = _doc(db, root.name, "garbage.pdf")
        assert doc.file_hash == _file_hash(path)
        embeds1, chats1 = len(llm.calls), len(llm.chat_calls)
        content1, digest1, indexed1 = doc.content, doc.content_hash, doc.indexed_at

        second = asyncio.run(import_sources([root], llm, session=db))
        assert (second.added, second.updated, second.unchanged) == (0, 0, 1)
        assert second.pdfs_screenshotted == 0  # the gate never re-fired
        # THE pin: the fallback vision did NOT re-fire.
        assert llm.vision_calls == 2, "fast path: no re-issued vision call"
        assert len(llm.chat_calls) == chats1
        assert len(llm.calls) == embeds1, "fast path: no re-embed"

        db.expire_all()
        doc2 = _doc(db, root.name, "garbage.pdf")
        assert doc2.content == content1  # content row byte-identical
        assert doc2.content_hash == digest1
        assert doc2.indexed_at == indexed1
    finally:
        _cleanup_source(db, root.name)


def test_pictorial_pdf_unchanged_second_sync_costs_zero_vision(
    db, tmp_path: Path
) -> None:
    """Pictorial PDF class (the IKEA assembly-instruction class,
    TODO L5): a sparse-page PDF indexed with the pictorial pass ON
    (six vision calls); re-run → ``unchanged == 1``, ZERO new llm
    calls — the pictorial pass did NOT re-fire (no re-render, no
    vision), and the content is byte-stable."""
    root = tmp_path / "fastikea"
    root.mkdir()
    path = root / "ikea.pdf"
    _build_ikea_like_pdf(path)
    llm = _PictorialLLM()
    llm.settings = _settings(tmp_path)  # pdf_pictorial ON (code default)
    try:
        first = asyncio.run(import_sources([root], llm, session=db))
        assert (first.added, first.pdfs_pictorial, first.pdf_pictorial_pages) == (1, 1, 6)
        assert llm.pictorial_calls == 6  # one per candidate — the first index
        doc = _doc(db, root.name, "ikea.pdf")
        assert doc.file_hash == _file_hash(path)
        embeds1, chats1 = len(llm.calls), len(llm.chat_calls)
        content1, digest1 = doc.content, doc.content_hash

        second = asyncio.run(import_sources([root], llm, session=db))
        assert (second.added, second.updated, second.unchanged) == (0, 0, 1)
        assert second.pdfs_pictorial == 0  # the pass did not re-fire
        # THE pin: the pictorial vision did NOT re-fire.
        assert llm.pictorial_calls == 6, "fast path: no re-issued vision call"
        assert len(llm.chat_calls) == chats1
        assert len(llm.calls) == embeds1, "fast path: no re-embed"

        db.expire_all()
        doc2 = _doc(db, root.name, "ikea.pdf")
        assert doc2.content == content1  # byte-stable content
        assert doc2.content_hash == digest1
    finally:
        _cleanup_source(db, root.name)


def test_image_unchanged_second_sync_costs_zero_llm(db, tmp_path: Path) -> None:
    """Image class: a standalone image indexed through the vision
    describe (one chat call); re-run → ``unchanged == 1``, ZERO new
    llm calls — the describe did NOT re-fire, no re-embed, and the
    row (content + ``indexed_at``) is untouched."""
    root = tmp_path / "fastimg"
    root.mkdir()
    path = root / "pic.png"
    path.write_bytes(_png_bytes(4, 4, (0.2, 0.4, 0.6)))
    llm = _VisionLLM()
    llm.settings = Settings(
        _env_file=None,  # pyright: ignore[reportCallIssue] (kwarg exists at runtime)
        images=True,
        image_dir=str(tmp_path / "images"),
    )
    try:
        first = asyncio.run(import_sources([root], llm, session=db))
        assert (first.added, first.images_failed) == (1, 0)
        doc = _doc(db, root.name, "pic.png")
        assert doc.file_hash == _file_hash(path)
        assert doc.content == _VisionLLM.DESCRIPTION
        embeds1, chats1 = len(llm.calls), len(llm.chat_calls)
        indexed1 = doc.indexed_at

        second = asyncio.run(import_sources([root], llm, session=db))
        assert (second.added, second.updated, second.unchanged) == (0, 0, 1)
        # THE pin: no re-describe, no re-embed.
        assert len(llm.chat_calls) == chats1, "fast path: no re-describe"
        assert len(llm.calls) == embeds1, "fast path: no re-embed"

        db.expire_all()
        doc2 = _doc(db, root.name, "pic.png")
        assert doc2.content == _VisionLLM.DESCRIPTION
        assert doc2.indexed_at == indexed1
    finally:
        _cleanup_source(db, root.name)


# ---------- changed files: full re-index + re-stamp ---------------------------


def test_changed_file_reindexes_and_re_stamps(db, tmp_path: Path) -> None:
    """A changed file (one byte flipped in a text file) re-indexes
    FULLY — the llm calls fire again (embed + summary) — and
    ``file_hash`` is RE-STAMPED with the new raw-bytes digest;
    ``content_hash`` moves with the content."""
    root = tmp_path / "fastchange"
    root.mkdir()
    path = root / "note.md"
    path.write_text("# Note\n\nv1\n")
    llm = FakeEmbedder()
    try:
        first = asyncio.run(import_sources([root], llm, session=db))
        assert first.added == 1
        doc = _doc(db, root.name, "note.md")
        hash1, content_digest1 = doc.file_hash, doc.content_hash
        assert hash1 == _file_hash(path)
        embeds1, chats1 = len(llm.calls), len(llm.chat_calls)

        path.write_text("# Note\n\nv2\n")  # the bytes changed
        second = asyncio.run(import_sources([root], llm, session=db))
        assert (second.added, second.updated, second.unchanged) == (0, 1, 0)
        # The full pipeline re-fired.
        assert len(llm.calls) > embeds1, "a changed file re-embeds"
        assert len(llm.chat_calls) > chats1, "a changed file re-summarizes"

        db.expire_all()
        doc2 = _doc(db, root.name, "note.md")
        assert doc2.file_hash == _file_hash(path)  # re-stamped
        assert doc2.file_hash != hash1
        assert doc2.content_hash != content_digest1
    finally:
        _cleanup_source(db, root.name)


def test_pdf_bytes_changed_content_same_stamps_without_content_change(
    db, tmp_path: Path
) -> None:
    """The bytes-changed/content-same edge (a PDF metadata edit with
    identical extracted text): the fast path does NOT fire (the raw
    bytes moved), the branch re-runs its pipeline (the fallback's
    vision re-issues — the honest cost of a byte change), but the
    CONTENT digest matches — the row is UNCHANGED (no content change,
    no re-embed), and the shared helper STAMPS the new hash, so the
    NEXT sync takes the fast path (zero further vision)."""
    root = tmp_path / "fastmeta"
    root.mkdir()
    path = root / "meta.pdf"
    _build_garbage_pdf(path)
    llm = _PageVisionLLM()
    llm.settings = _pdf_settings(tmp_path)
    try:
        first = asyncio.run(import_sources([root], llm, session=db))
        assert (first.added, first.pdfs_screenshotted) == (1, 1)
        doc = _doc(db, root.name, "meta.pdf")
        content1, digest1 = doc.content, doc.content_hash
        hash1 = doc.file_hash
        assert llm.vision_calls == 2
        embeds1 = len(llm.calls)

        # A metadata-only edit: the raw bytes change, the extracted
        # text (and the rendered pages) do not.
        with fitz.open(str(path)) as d:
            d.set_metadata({"title": "metadata-only edit"})
            d.saveIncr()
        new_hash = _file_hash(path)
        assert new_hash != hash1  # the bytes really moved

        second = asyncio.run(import_sources([root], llm, session=db))
        # The branch re-ran (gate re-fired, vision re-issued — the byte
        # change's cost) but the content digest matched: the row is
        # unchanged, no content change, no re-embed.
        assert (second.added, second.updated, second.unchanged) == (0, 0, 1)
        assert second.pdfs_screenshotted == 1  # the branch re-ran its gate
        assert llm.vision_calls == 4  # the fallback re-issued (bytes changed)
        assert len(llm.calls) == embeds1, "content-same: no re-embed"

        db.expire_all()
        doc2 = _doc(db, root.name, "meta.pdf")
        assert doc2.content == content1  # content byte-identical
        assert doc2.content_hash == digest1
        assert doc2.file_hash == new_hash  # STAMPED by the shared helper

        # The NEXT sync is fast (the stamp landed): zero further vision.
        third = asyncio.run(import_sources([root], llm, session=db))
        assert (third.added, third.updated, third.unchanged) == (0, 0, 1)
        assert third.pdfs_screenshotted == 0
        assert llm.vision_calls == 4, "the third sync is the fast path"
        assert len(llm.calls) == embeds1
    finally:
        _cleanup_source(db, root.name)


# ---------- pre-phase rows (NULL file_hash) -----------------------------------


def test_pre_phase_null_file_hash_runs_branch_path_and_stamps(
    db, tmp_path: Path
) -> None:
    """A row that predates the column (``file_hash`` NULL) + an
    unchanged file: the fast path cannot fire, so the branch's
    content-digest path runs (today's behavior — the digest matches,
    no re-embed), and the shared helper STAMPS the hash — the THIRD
    sync then takes the fast path (proof the stamp landed)."""
    root = tmp_path / "fastlegacy"
    root.mkdir()
    path = root / "note.md"
    path.write_text("# Note\n\nmarkdown body\n")
    llm = FakeEmbedder()
    try:
        asyncio.run(import_sources([root], llm, session=db))
        doc = _doc(db, root.name, "note.md")
        doc.file_hash = None  # simulate the pre-phase row
        db.commit()

        embeds1, chats1 = len(llm.calls), len(llm.chat_calls)
        second = asyncio.run(import_sources([root], llm, session=db))
        assert (second.added, second.updated, second.unchanged) == (0, 0, 1)
        # The content digest matched — no re-embed, no re-summary.
        assert len(llm.calls) == embeds1
        assert len(llm.chat_calls) == chats1

        db.expire_all()
        doc2 = _doc(db, root.name, "note.md")
        assert doc2.file_hash == _file_hash(path)  # stamped by the helper

        # The third sync takes the FAST PATH (the stamp is committed):
        third = asyncio.run(import_sources([root], llm, session=db))
        assert third.unchanged == 1
        assert len(llm.calls) == embeds1
        assert len(llm.chat_calls) == chats1
    finally:
        _cleanup_source(db, root.name)


def test_pre_phase_null_file_hash_image_runs_branch_and_stamps(
    db, tmp_path: Path
) -> None:
    """The image flavour of the branch path: a pre-phase row
    (``file_hash`` NULL) + an unchanged image — the fast path cannot
    fire, so the image branch's content-digest check runs (the raw
    BYTES digest — the image's content identity — matches BEFORE the
    expensive describe, so no re-describe, no re-embed), the row is
    unchanged, and the shared helper STAMPS the hash — the next sync
    is fast.
    """
    root = tmp_path / "fastimglegacy"
    root.mkdir()
    path = root / "pic.png"
    path.write_bytes(_png_bytes(4, 4, (0.2, 0.4, 0.6)))
    llm = _VisionLLM()
    llm.settings = Settings(
        _env_file=None,  # pyright: ignore[reportCallIssue] (kwarg exists at runtime)
        images=True,
        image_dir=str(tmp_path / "images"),
    )
    try:
        asyncio.run(import_sources([root], llm, session=db))
        doc = _doc(db, root.name, "pic.png")
        doc.file_hash = None  # simulate the pre-phase row
        db.commit()

        chats1 = len(llm.chat_calls)
        embeds1 = len(llm.calls)
        second = asyncio.run(import_sources([root], llm, session=db))
        assert (second.added, second.updated, second.unchanged) == (0, 0, 1)
        # The branch's byte-digest check matched before the describe —
        # zero llm work of any kind.
        assert len(llm.chat_calls) == chats1
        assert len(llm.calls) == embeds1, "content-same: no re-embed"

        db.expire_all()
        doc2 = _doc(db, root.name, "pic.png")
        assert doc2.file_hash == _file_hash(path)  # stamped by the helper
        assert doc2.content == _VisionLLM.DESCRIPTION

        # The next sync takes the fast path (zero further calls):
        third = asyncio.run(import_sources([root], llm, session=db))
        assert third.unchanged == 1
        assert len(llm.chat_calls) == chats1
        assert len(llm.calls) == embeds1
    finally:
        _cleanup_source(db, root.name)


# ---------- phase 163 (task 03): the deterministic PDF strip re-gate ----------
#
# The live defect (owner report, 2026-10-07): the `Letumdei` row carried
# `page_count = 36` over ZERO files on disk (the pre-phase-163 persist
# deleted the old copies BEFORE rendering, outside the DB transaction —
# a later per-file rollback restored the stale count over an empty
# directory; an external wipe of `pdf_page_dir` lands the same state).
# The phase-147 fast path believed "the copies persist" and never
# re-rendered, so the strip 404'd forever. `_regate_pdf_strip` now runs
# at the tail of `_unchanged_doc`: the changed-path gate mirrored from
# the current bytes + settings (ZERO vision/LLM calls), missing copies
# healed, a stale strip cleared.

#: The PNG signature (a healed copy is a real render, not a placeholder).
PNG_MAGIC = b"\x89PNG\r\n\x1a\n"


def _regate_settings(tmp_path: Path, **kwargs: Any) -> Settings:
    """Settings for the re-gate matrix: the ``pdf`` token admitted, the
    phase-140 knobs at their CODE DEFAULTS (``pdf_pictorial`` ON), and
    the strip's home ``pdf_page_dir`` pinned under *tmp_path* (never the
    real home)."""
    kwargs.setdefault("_env_file", None)
    kwargs.setdefault("import_extensions", "md,txt,pdf")
    kwargs.setdefault("images", False)
    kwargs.setdefault("image_dir", str(tmp_path / "images"))
    kwargs.setdefault("pdf_page_dir", str(tmp_path / "pages"))
    return Settings(**kwargs)  # pyright: ignore[reportCallIssue]


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


def _strip_copies(doc: Document, page_dir: Path, count: int) -> list[Path]:
    return [page_dir / f"{doc.id}-p{n}.png" for n in range(1, count + 1)]


def test_regate_heals_deleted_copies_of_a_pictorial_strip(
    db, tmp_path: Path, caplog: pytest.LogCaptureFixture
) -> None:
    """THE live-loss reproduction (heal arm): a pictorial-class PDF is
    indexed (``page_count`` set, copies written), its copies are deleted
    out from under the row (the exact live state — mode NULL on a
    pre-phase row), and an UNCHANGED re-sync restores every file with
    real PNG bytes, stamps the mode ``additive``, keeps ``page_count``
    (no drift), burns ZERO llm/vision calls, and logs the one greppable
    INFO line. The NULL-mode stamp is SILENT (no ``mode`` segment — it
    rides the line only when work followed)."""
    root = tmp_path / "rgheal"
    root.mkdir()
    _build_ikea_like_pdf(root / "ikea.pdf")
    llm = _PictorialLLM()
    llm.settings = _regate_settings(tmp_path)
    try:
        asyncio.run(import_sources([root], llm, session=db))
        doc = _doc(db, "rgheal", "ikea.pdf")
        assert (doc.page_count, doc.page_strip_mode) == (10, "additive")
        page_dir = _pages_dir(tmp_path)
        doc.page_strip_mode = None  # the pre-phase row (the live shape)
        db.commit()
        for path in _strip_copies(doc, page_dir, 10):
            path.unlink()
        assert not list(page_dir.glob(f"{doc.id}-*.png"))
        embeds1, chats1, vision1 = len(llm.calls), len(llm.chat_calls), llm.pictorial_calls

        with caplog.at_level(logging.INFO, logger="app.importer"):
            second = asyncio.run(import_sources([root], llm, session=db))

        assert (second.added, second.updated, second.unchanged) == (0, 0, 1)
        # ZERO llm work of any kind (the re-gate is deterministic, A5).
        assert len(llm.calls) == embeds1, "the re-gate never re-embeds"
        assert len(llm.chat_calls) == chats1, "the re-gate never chats"
        assert llm.pictorial_calls == vision1, "the re-gate issues ZERO vision calls"

        db.expire_all()
        doc2 = _doc(db, "rgheal", "ikea.pdf")
        assert doc2.page_count == 10, "no drift — the count is untouched"
        assert doc2.page_strip_mode == "additive", "the mode is stamped"
        for path in _strip_copies(doc2, page_dir, 10):
            assert path.is_file(), f"copy {path.name} was not healed"
            assert path.read_bytes()[:8] == PNG_MAGIC, f"{path.name} is not a PNG"
        assert not list(page_dir.glob("*.tmp")), "no .tmp orphan from the heal"
        messages = [r.getMessage() for r in caplog.records]
        assert (
            "import: PDF strip re-gate source=rgheal path=ikea.pdf — "
            "healed 10 pages"
        ) in messages, messages
        assert not any("mode " in m for m in messages if "re-gate" in m), (
            "a NULL->stamp mode re-stamp is silent-beneficial " + str(messages)
        )
    finally:
        _cleanup_source(db, "rgheal")


def test_regate_heals_a_fallback_strip_as_primary(
    db, tmp_path: Path, caplog: pytest.LogCaptureFixture
) -> None:
    """Heal arm, the FALLBACK class (phase 133's unusable text layer):
    the copies of a garbage-layer PDF's strip are restored on an
    unchanged re-sync and the mode lands ``primary`` (the strip IS the
    view) — and the phase-133 fallback vision does NOT re-fire (the
    class is re-derived from the bytes, no vision call)."""
    root = tmp_path / "rgfall"
    root.mkdir()
    _build_garbage_pdf(root / "garbage.pdf")  # 2 pages
    llm = _PageVisionLLM()
    llm.settings = _regate_settings(tmp_path)
    try:
        asyncio.run(import_sources([root], llm, session=db))
        doc = _doc(db, "rgfall", "garbage.pdf")
        assert (doc.page_count, doc.page_strip_mode) == (2, "primary")
        page_dir = _pages_dir(tmp_path)
        for path in _strip_copies(doc, page_dir, 2):
            path.unlink()
        vision1, chats1, embeds1 = llm.vision_calls, len(llm.chat_calls), len(llm.calls)

        with caplog.at_level(logging.INFO, logger="app.importer"):
            second = asyncio.run(import_sources([root], llm, session=db))

        assert (second.added, second.updated, second.unchanged) == (0, 0, 1)
        assert second.pdfs_screenshotted == 0, "the fallback gate did not re-run"
        assert llm.vision_calls == vision1, "ZERO vision calls on the re-gate"
        assert len(llm.chat_calls) == chats1 and len(llm.calls) == embeds1

        db.expire_all()
        doc2 = _doc(db, "rgfall", "garbage.pdf")
        assert doc2.page_strip_mode == "primary", "the unusable layer is primary"
        assert doc2.page_count == 2
        for path in _strip_copies(doc2, page_dir, 2):
            assert path.is_file() and path.read_bytes()[:8] == PNG_MAGIC
        assert any(
            "healed 2 pages" in m and "re-gate" in m for m in caplog.messages
        ), caplog.messages
    finally:
        _cleanup_source(db, "rgfall")


def test_regate_noop_when_the_strip_is_healthy(
    db, tmp_path: Path, caplog: pytest.LogCaptureFixture
) -> None:
    """No-op arm (the IKEA-class doc whose copies all stand): an
    unchanged re-sync writes NOTHING (every copy's mtime and bytes are
    untouched), renders NOTHING, and logs NO re-gate line — the INFO is
    never per-unchanged-doc — while the pre-existing ``import:
    unchanged`` line still fires."""
    root = tmp_path / "rgnoop"
    root.mkdir()
    _build_ikea_like_pdf(root / "ikea.pdf")
    llm = _PictorialLLM()
    llm.settings = _regate_settings(tmp_path)
    try:
        asyncio.run(import_sources([root], llm, session=db))
        doc = _doc(db, "rgnoop", "ikea.pdf")
        page_dir = _pages_dir(tmp_path)
        before = {
            p: (p.stat().st_mtime_ns, p.read_bytes())
            for p in _strip_copies(doc, page_dir, 10)
        }
        assert len(before) == 10

        with caplog.at_level(logging.INFO, logger="app.importer"):
            second = asyncio.run(import_sources([root], llm, session=db))

        assert (second.added, second.updated, second.unchanged) == (0, 0, 1)
        db.expire_all()
        doc2 = _doc(db, "rgnoop", "ikea.pdf")
        assert (doc2.page_count, doc2.page_strip_mode) == (10, "additive")
        after = {
            p: (p.stat().st_mtime_ns, p.read_bytes())
            for p in _strip_copies(doc2, page_dir, 10)
        }
        assert after == before, "a healthy strip is not written at all"
        messages = [r.getMessage() for r in caplog.records]
        assert any(m.startswith("import: unchanged ") for m in messages), messages
        assert not any("re-gate" in m for m in messages), (
            f"the re-gate logged without acting: {messages}"
        )
    finally:
        _cleanup_source(db, "rgnoop")


def test_regate_restamps_a_changed_mode(
    db, tmp_path: Path, caplog: pytest.LogCaptureFixture
) -> None:
    """Mode re-stamp (the class flip): a row whose stamped mode no
    longer matches the gate's class (here a pictorial-class row carrying
    ``primary`` — a legacy/mis-stamped row) is re-stamped to
    ``additive``, and because the mode was ALREADY SET the change is
    reported (``mode primary->additive``) — the only thing the re-gate
    did, so that is the whole INFO line (no copies were missing, none
    were written)."""
    root = tmp_path / "rgmode"
    root.mkdir()
    _build_ikea_like_pdf(root / "ikea.pdf")
    llm = _PictorialLLM()
    llm.settings = _regate_settings(tmp_path)
    try:
        asyncio.run(import_sources([root], llm, session=db))
        doc = _doc(db, "rgmode", "ikea.pdf")
        assert doc.page_strip_mode == "additive"
        doc.page_strip_mode = "primary"  # the mis-stamp the re-gate fixes
        db.commit()
        page_dir = _pages_dir(tmp_path)
        before = {
            p: p.stat().st_mtime_ns for p in _strip_copies(doc, page_dir, 10)
        }

        with caplog.at_level(logging.INFO, logger="app.importer"):
            second = asyncio.run(import_sources([root], llm, session=db))

        assert (second.added, second.unchanged) == (0, 1)
        db.expire_all()
        doc2 = _doc(db, "rgmode", "ikea.pdf")
        assert doc2.page_strip_mode == "additive", "the class is re-stamped"
        assert doc2.page_count == 10
        assert {
            p: p.stat().st_mtime_ns for p in _strip_copies(doc2, page_dir, 10)
        } == before, "a mode fix writes no files"
        assert (
            "import: PDF strip re-gate source=rgmode path=ikea.pdf — "
            "mode primary->additive"
        ) in [r.getMessage() for r in caplog.records], caplog.messages
    finally:
        _cleanup_source(db, "rgmode")


def test_regate_clears_a_strip_that_no_longer_qualifies(
    db, tmp_path: Path, caplog: pytest.LogCaptureFixture
) -> None:
    """REPAIR arm: a row claims a strip (``page_count`` set, copies on
    disk) on a PDF whose GATE now says ineligible (a clean, dense text
    layer — zero sparse pages). The unchanged re-sync clears BOTH
    columns and deletes the copies (the strip must not linger over a
    document the viewer shows as text) + the ``cleared stale strip``
    INFO line, with zero llm calls."""
    root = tmp_path / "rgclear"
    root.mkdir()
    pdf = root / "clean.pdf"
    _build_dense_pdf(pdf)  # usable layer, zero candidates — never qualifies
    llm = _PictorialLLM()
    llm.settings = _regate_settings(tmp_path)
    try:
        asyncio.run(import_sources([root], llm, session=db))
        doc = _doc(db, "rgclear", "clean.pdf")
        assert (doc.page_count, doc.page_strip_mode) == (None, None)
        # The stale state the re-gate must repair (a strip stamped by an
        # older gate/settings, copies still on disk).
        page_dir = _pages_dir(tmp_path)
        page_dir.mkdir(parents=True, exist_ok=True)
        for path in _strip_copies(doc, page_dir, 2):
            path.write_bytes(b"STALE-COPY")
        doc.page_count = 2
        doc.page_strip_mode = "additive"
        db.commit()
        embeds1, chats1 = len(llm.calls), len(llm.chat_calls)

        with caplog.at_level(logging.INFO, logger="app.importer"):
            second = asyncio.run(import_sources([root], llm, session=db))

        assert (second.added, second.updated, second.unchanged) == (0, 0, 1)
        assert len(llm.calls) == embeds1 and len(llm.chat_calls) == chats1

        db.expire_all()
        doc2 = _doc(db, "rgclear", "clean.pdf")
        assert doc2.page_count is None, "the stale strip is cleared"
        assert doc2.page_strip_mode is None, "the class is cleared with it"
        assert not list(page_dir.glob(f"{doc2.id}-p*.png")), "the copies are deleted"
        assert (
            "import: PDF strip re-gate source=rgclear path=clean.pdf — "
            "cleared stale strip"
        ) in [r.getMessage() for r in caplog.records], caplog.messages
    finally:
        _cleanup_source(db, "rgclear")


def test_regate_kill_switch_clears_the_pictorial_strip(
    db, tmp_path: Path, caplog: pytest.LogCaptureFixture
) -> None:
    """REPAIR arm via the kill switch (``pdf_pictorial=False``): the
    changed-path semantics mirrored — with the pass off a sparse-page
    PDF is NOT eligible, so its existing strip is cleared with its
    copies on the next (unchanged) sync. The bytes never moved: the
    setting alone decided."""
    root = tmp_path / "rgkill"
    root.mkdir()
    _build_ikea_like_pdf(root / "ikea.pdf")
    llm = _PictorialLLM()
    llm.settings = _regate_settings(tmp_path)
    try:
        asyncio.run(import_sources([root], llm, session=db))
        doc = _doc(db, "rgkill", "ikea.pdf")
        assert (doc.page_count, doc.page_strip_mode) == (10, "additive")
        page_dir = _pages_dir(tmp_path)
        assert len(list(page_dir.glob(f"{doc.id}-p*.png"))) == 10
        vision1 = llm.pictorial_calls

        llm.settings = _regate_settings(tmp_path, pdf_pictorial=False)
        with caplog.at_level(logging.INFO, logger="app.importer"):
            second = asyncio.run(import_sources([root], llm, session=db))

        assert (second.added, second.updated, second.unchanged) == (0, 0, 1)
        assert llm.pictorial_calls == vision1, "ZERO vision calls (the pass is off)"
        db.expire_all()
        doc2 = _doc(db, "rgkill", "ikea.pdf")
        assert doc2.page_count is None and doc2.page_strip_mode is None
        assert not list(page_dir.glob(f"{doc2.id}-p*.png")), "the copies went too"
        assert any(
            "cleared stale strip" in m for m in caplog.messages
        ), caplog.messages
    finally:
        _cleanup_source(db, "rgkill")


def test_regate_render_exception_is_fail_soft(
    db,
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
    caplog: pytest.LogCaptureFixture,
) -> None:
    """Fail-soft: the heal's render RAISES — the sync completes
    (``unchanged``, zero errors), the row keeps its prior state
    (``page_count`` + mode untouched — the re-gate never clears a strip
    on a failure), no copies are touched, and the WARNING is the only
    signal."""
    root = tmp_path / "rgboom"
    root.mkdir()
    _build_ikea_like_pdf(root / "ikea.pdf")
    llm = _PictorialLLM()
    llm.settings = _regate_settings(tmp_path)
    try:
        asyncio.run(import_sources([root], llm, session=db))
        doc = _doc(db, "rgboom", "ikea.pdf")
        page_dir = _pages_dir(tmp_path)
        for path in _strip_copies(doc, page_dir, 10):
            path.unlink()

        def boom(*args: object, **kwargs: object) -> list[bytes]:
            raise RuntimeError("simulated render failure (test sentinel)")

        monkeypatch.setattr(importer, "render_pdf_pages", boom)
        with caplog.at_level(logging.WARNING, logger="app.importer"):
            second = asyncio.run(import_sources([root], llm, session=db))

        assert (second.added, second.updated, second.unchanged, second.errors) == (
            0, 0, 1, 0,
        )
        db.expire_all()
        doc2 = _doc(db, "rgboom", "ikea.pdf")
        assert (doc2.page_count, doc2.page_strip_mode) == (10, "additive"), (
            "the row keeps its pre-re-gate state"
        )
        assert not list(page_dir.glob(f"{doc.id}-p*.png")), "no copies were touched"
        assert any(
            "PDF strip re-gate failed" in m and "test sentinel" in m
            for m in caplog.messages
        ), caplog.messages
    finally:
        _cleanup_source(db, "rgboom")


def test_regate_empty_render_keeps_the_row(
    db,
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
    caplog: pytest.LogCaptureFixture,
) -> None:
    """Fail-soft on the no-pages render (unopenable right after the fast
    path hashed the file — pathological): the row is kept EXACTLY as it
    was (never cleared on a transient), nothing is written, and the
    WARNING names the no-pages state."""
    root = tmp_path / "rgempty"
    root.mkdir()
    _build_ikea_like_pdf(root / "ikea.pdf")
    llm = _PictorialLLM()
    llm.settings = _regate_settings(tmp_path)
    try:
        asyncio.run(import_sources([root], llm, session=db))
        doc = _doc(db, "rgempty", "ikea.pdf")
        page_dir = _pages_dir(tmp_path)
        for path in _strip_copies(doc, page_dir, 10):
            path.unlink()
        monkeypatch.setattr(importer, "render_pdf_pages", lambda *a, **k: [])

        with caplog.at_level(logging.WARNING, logger="app.importer"):
            second = asyncio.run(import_sources([root], llm, session=db))

        assert (second.added, second.unchanged, second.errors) == (0, 1, 0)
        db.expire_all()
        doc2 = _doc(db, "rgempty", "ikea.pdf")
        assert (doc2.page_count, doc2.page_strip_mode) == (10, "additive")
        assert not page_dir.exists() or not list(page_dir.glob(f"{doc.id}-*"))
        assert any(
            "re-render produced no pages (state kept)" in m
            for m in caplog.messages
        ), caplog.messages
    finally:
        _cleanup_source(db, "rgempty")


def test_regate_reports_a_page_count_drift_on_the_heal(
    db,
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
    caplog: pytest.LogCaptureFixture,
) -> None:
    """Drift (pathological on unchanged bytes, so it is LOUD): the heal
    renders 6 pages for a row that claimed 10 — ``page_count`` follows
    the fresh render, the ``p7..p10`` tail beyond it is deleted, and a
    WARNING names both counts beside the ``healed`` INFO line."""
    root = tmp_path / "rgdrift"
    root.mkdir()
    _build_ikea_like_pdf(root / "ikea.pdf")
    llm = _PictorialLLM()
    llm.settings = _regate_settings(tmp_path)
    try:
        asyncio.run(import_sources([root], llm, session=db))
        doc = _doc(db, "rgdrift", "ikea.pdf")
        page_dir = _pages_dir(tmp_path)
        for path in _strip_copies(doc, page_dir, 10):
            path.unlink()
        monkeypatch.setattr(
            importer,
            "render_pdf_pages",
            lambda *a, **k: [PNG_MAGIC + f"{i}".encode() for i in range(6)],
        )

        with caplog.at_level(logging.INFO, logger="app.importer"):
            second = asyncio.run(import_sources([root], llm, session=db))

        assert (second.added, second.unchanged) == (0, 1)
        db.expire_all()
        doc2 = _doc(db, "rgdrift", "ikea.pdf")
        assert doc2.page_count == 6, "the count follows the fresh render"
        for n in range(1, 7):
            assert (page_dir / f"{doc2.id}-p{n}.png").is_file()
        for n in range(7, 11):
            assert not (page_dir / f"{doc2.id}-p{n}.png").exists(), f"p{n} leaked"
        messages = [r.getMessage() for r in caplog.records]
        assert any("page_count drift 10->6" in m for m in messages), messages
        assert any("healed 10 pages" in m for m in messages), messages
    finally:
        monkeypatch.undo()  # pyright: ignore[reportUnknownMemberType]
        _cleanup_source(db, "rgdrift")


def test_regate_fires_on_the_content_digest_hit_path(
    db, tmp_path: Path, caplog: pytest.LogCaptureFixture
) -> None:
    """The SECOND unchanged path (the PDF branch's content-digest hit —
    a pre-phase row with ``file_hash`` NULL): the branch re-runs its own
    gate (the byte-change's honest cost) and lands unchanged, and the
    shared helper's tail still heals the strip deleted out from under
    the row — the re-gate is not fast-path-only."""
    root = tmp_path / "rgbranch"
    root.mkdir()
    _build_ikea_like_pdf(root / "ikea.pdf")
    llm = _PictorialLLM()
    llm.settings = _regate_settings(tmp_path)
    try:
        asyncio.run(import_sources([root], llm, session=db))
        doc = _doc(db, "rgbranch", "ikea.pdf")
        doc.file_hash = None  # the pre-phase row: the fast path cannot fire
        db.commit()
        page_dir = _pages_dir(tmp_path)
        for path in _strip_copies(doc, page_dir, 10):
            path.unlink()

        with caplog.at_level(logging.INFO, logger="app.importer"):
            second = asyncio.run(import_sources([root], llm, session=db))

        assert (second.added, second.updated, second.unchanged) == (0, 0, 1)
        db.expire_all()
        doc2 = _doc(db, "rgbranch", "ikea.pdf")
        assert doc2.file_hash is not None, "the helper stamped the hash"
        for path in _strip_copies(doc2, page_dir, 10):
            assert path.is_file(), f"copy {path.name} was not healed"
        assert any("healed 10 pages" in m for m in caplog.messages), caplog.messages
    finally:
        _cleanup_source(db, "rgbranch")


def test_regate_guard_costs_a_non_strip_file_nothing(
    db, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    """The GUARD (A5): a non-PDF unchanged file and a PDF whose
    ``page_count`` is NULL reach the tail of ``_unchanged_doc`` and
    ``extract_pdf_text`` is NEVER called for either — every non-strip
    file's fast path keeps its zero-extra-read contract (one sync of
    each class, then the spy covers the second sync only)."""
    root = tmp_path / "rgguard"
    root.mkdir()
    (root / "note.md").write_text("# Note\n\nplain text body\n")
    _build_dense_pdf(root / "clean.pdf")  # usable layer, no strip
    llm = _PictorialLLM()
    llm.settings = _regate_settings(tmp_path)
    seen: list[str] = []
    real_extract = importer.extract_pdf_text

    def spy(path: Path) -> str:
        seen.append(str(path))
        return real_extract(path)

    try:
        asyncio.run(import_sources([root], llm, session=db))
        monkeypatch.setattr(importer, "extract_pdf_text", spy)
        second = asyncio.run(import_sources([root], llm, session=db))
        assert (second.added, second.updated, second.unchanged) == (0, 0, 2)
        assert seen == [], f"the guard let a PDF parse through: {seen}"
    finally:
        monkeypatch.undo()  # pyright: ignore[reportUnknownMemberType]
        _cleanup_source(db, "rgguard")


# ---------- the preserved semantics on the fast path --------------------------


def test_manual_date_row_fast_path_no_date_write(db, tmp_path: Path) -> None:
    """D1 on the fast path: an unchanged file carrying the owner's
    correction (``created_at_manual``) gets NO date write — the moved
    mtime does not refresh it, ``dates_updated`` stays 0, the
    correction + the flag survive, and the row's stamp is intact
    (zero llm calls)."""
    root = tmp_path / "fastmanual"
    root.mkdir()
    path = root / "note.md"
    path.write_text("# Note\n\ncontent for the KB\n")
    llm = FakeEmbedder()
    correction = datetime(2023, 5, 5, 9, 30, 0, tzinfo=UTC)
    try:
        asyncio.run(import_sources([root], llm, session=db))
        doc = _doc(db, root.name, "note.md")
        doc.created_at = correction
        doc.created_at_manual = True
        db.commit()
        _utime(path, OLD_2021)  # the source moved to a DIFFERENT date
        embeds1, chats1 = len(llm.calls), len(llm.chat_calls)

        second = asyncio.run(import_sources([root], llm, session=db))
        assert (second.added, second.updated, second.unchanged) == (0, 0, 1)
        assert second.dates_updated == 0  # the manual lock withheld it
        assert len(llm.calls) == embeds1 and len(llm.chat_calls) == chats1

        db.expire_all()
        doc2 = _doc(db, root.name, "note.md")
        assert doc2.created_at == correction  # the correction survived
        assert doc2.created_at_manual is True  # the flag intact
        assert doc2.file_hash == _file_hash(path)  # the stamp intact
    finally:
        _cleanup_source(db, root.name)


def test_manual_date_null_hash_row_stamps_without_date_write(
    db, tmp_path: Path
) -> None:
    """D1 on the BRANCH path (a pre-phase NULL-hash row that also
    carries the manual correction): the shared helper's stamp is the
    only write — it COMMITS (so the next sync is fast even for a
    manual-date row) — while the DATE stays the manual value
    (``dates_updated`` 0, the flag intact)."""
    root = tmp_path / "fastmanualnull"
    root.mkdir()
    path = root / "note.md"
    path.write_text("# Note\n\ncontent for the KB\n")
    llm = FakeEmbedder()
    correction = datetime(2023, 5, 5, 9, 30, 0, tzinfo=UTC)
    try:
        asyncio.run(import_sources([root], llm, session=db))
        doc = _doc(db, root.name, "note.md")
        doc.created_at = correction
        doc.created_at_manual = True
        doc.file_hash = None  # pre-phase row
        db.commit()
        _utime(path, OLD_2021)  # a refresh is DUE — the lock must withhold it
        embeds1, chats1 = len(llm.calls), len(llm.chat_calls)

        second = asyncio.run(import_sources([root], llm, session=db))
        assert (second.added, second.updated, second.unchanged) == (0, 0, 1)
        assert second.dates_updated == 0
        # No summary work of any kind on this run (run 1's summary
        # count is the baseline).
        assert len(llm.chat_calls) == chats1
        assert len(llm.calls) == embeds1

        db.expire_all()
        doc2 = _doc(db, root.name, "note.md")
        assert doc2.created_at == correction  # the correction survived
        assert doc2.created_at_manual is True
        assert doc2.file_hash == _file_hash(path)  # the stamp committed

        # The next sync is fast (the stamp landed despite the D1 lock):
        third = asyncio.run(import_sources([root], llm, session=db))
        assert third.unchanged == 1 and third.dates_updated == 0
        db.expire_all()
        assert _doc(db, root.name, "note.md").created_at == correction
    finally:
        _cleanup_source(db, root.name)


def test_unchanged_fast_path_date_refresh_may_go_older(
    db, tmp_path: Path, caplog: pytest.LogCaptureFixture
) -> None:
    """D4 on the fast path: an unchanged file whose mtime-derived date
    moved (OLDER — no monotonic guard) refreshes the stored date on
    the fast path: ``dates_updated == 1``, the date lands, and zero
    llm calls (the refresh is bookkeeping, not a re-index). The
    PRE-CHANGE log wordings are pinned verbatim (``import: unchanged``
    + ``import: date-refreshed``)."""
    root = tmp_path / "fastdate"
    root.mkdir()
    path = root / "note.md"
    path.write_text("# Note\n\ncontent for the KB\n")
    llm = FakeEmbedder()
    try:
        asyncio.run(import_sources([root], llm, session=db))
        _utime(path, OLD_2021)  # the source date moved; content identical
        embeds1, chats1 = len(llm.calls), len(llm.chat_calls)

        with caplog.at_level(logging.INFO, logger="app.importer"):
            second = asyncio.run(import_sources([root], llm, session=db))
        assert (second.added, second.updated, second.unchanged) == (0, 0, 1)
        assert second.dates_updated == 1  # the refresh ran (fast path)
        assert len(llm.calls) == embeds1 and len(llm.chat_calls) == chats1
        # The preserved wordings (the shared helper's log lines, verbatim
        # from the per-branch blocks).
        messages = [r.getMessage() for r in caplog.records]
        assert f"import: unchanged source={root.name} path=note.md" in messages
        assert any(
            m.startswith(f"import: date-refreshed source={root.name} path=note.md date=")
            for m in messages
        )

        db.expire_all()
        doc2 = _doc(db, root.name, "note.md")
        assert abs(doc2.created_at - OLD_2021) <= _TOL
    finally:
        _cleanup_source(db, root.name)


def test_null_summary_backfill_on_fast_path(db, tmp_path: Path) -> None:
    """Phase 118 (A2) via the fast path: an unchanged doc whose
    summary is still NULL (the stored content is intact) gets the
    summary-only backfill from the STORED ``doc.content`` — one
    ``is_summary`` chunk, no content re-embed, counted
    ``summary_backfilled`` (never ``summaries``), never
    added/updated/pruned. The ``lite`` call fires (the backfill's own
    work — the fast path's "zero llm calls" is "unless the backfill
    fires"); nothing else does."""
    root = tmp_path / "fastbfill"
    root.mkdir()
    (root / "note.md").write_text("# Note\n\nmarkdown body\n")
    llm = FakeEmbedder()
    try:
        asyncio.run(import_sources([root], llm, session=db))
        doc = _doc(db, root.name, "note.md")
        # Simulate the NULL-summary target row: the content stays, only
        # the summary + its chunk go away (the phase-118 convention).
        doc.summary = None
        for c in [c for c in doc.chunks if c.is_summary]:
            doc.chunks.remove(c)
        db.commit()
        assert doc.file_hash is not None  # the fast path can fire

        embeds1, chats1 = len(llm.calls), len(llm.chat_calls)
        second = asyncio.run(import_sources([root], llm, session=db))
        assert (second.added, second.updated, second.pruned) == (0, 0, 0)
        assert (second.unchanged, second.summary_backfilled, second.summaries) == (1, 1, 0)
        assert second.summary_errors == 0
        # Exactly the backfill's own work: one lite call, one embed
        # batch (the summary chunk only — no content re-embed).
        assert len(llm.chat_calls) == chats1 + 1
        assert len(llm.calls) == embeds1 + 1
        assert second.changed_paths == []  # no KB change recorded

        db.expire_all()
        doc2 = _doc(db, root.name, "note.md")
        assert doc2.summary is not None
        schunks = [c for c in doc2.chunks if c.is_summary]
        assert len(schunks) == 1 and schunks[0].position == -1
        assert schunks[0].content == doc2.summary
        content = [c for c in doc2.chunks if not c.is_summary]
        assert len(content) == 1 and content[0].embedding is not None
    finally:
        _cleanup_source(db, root.name)