"""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()``).

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

import pymupdf as fitz
import pytest
from sqlalchemy import select

from app.config import Settings
from app.models import Document
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_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)


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