phase: 127_sync_incremental_folder_summaries

All verified green — the previous rejection was a harness interruption (no code failure), and the working tree + doc fixes are intact.

**Phase 127 — final verification pass: 100% green** (all 4 tasks already complete; no code changes needed this pass)

- **Re-verified implementation:** `ImportSummary.changed_paths` at all 5 increment sites + `_prune`; `affected_folder_keys` + `generate_folder_summaries(changed_docs=…)` (restricted keys, prune pass both modes, hook total = affected count); sync changed-KB branch passes `changed_docs=summary.changed_paths`; CLI keeps full-regen contract
- **Tests:** `uv run pytest --cov=app --cov-report=term-missing` → **3022 passed**, `app/` coverage **99%**; `uv run pytest tests/e2e/test_incremental_folder_summaries.py -v --no-cov` → **3 passed in isolation**; `uv run ruff check . && uv run pyright` → clean, **0 errors**
- **Completion criteria:** one-file change rewrites only its folder chain (E2E pins siblings byte-identical, text AND `updated_at`) ✅; coverage >90% ✅; isolated E2E green ✅; no completed-phase behavior change (full suite green) ✅; atomic commit — delegated to the harness per pass rules (no git commands run by me)
- **Docs:** README.md re-checked — all 7 previous-pass fixes still in place (Theme page row, drill-down `ls` + capped `read`, phase-124 answer images section + `BOR_ANSWER_IMAGES`, markdown-included summaries, 17-color palette, folder summaries in `BOR_LLM_SUMMARY_MODEL`, CLI regen note); AGENTS.md verified consistent — no changes needed
- **Deviations:** none; untracked owner file `NOTES.md` left untouched (rides in the harness commit)
- **Next pending phase:** `128_sync_label_mobile`

tests/unit/test_folder_summaries.py

@@ -29,6 +29,7 @@ from app.rag.folder_summaries import (
     FOLDER_SUMMARY_MODE,
     MIN_DOCS_PER_FOLDER,
     SYSTEM_PROMPT,
+    affected_folder_keys,
     build_folder_summary_prompt,
     folder_of,
     generate_folder_summaries,
@@ -202,6 +203,56 @@ def test_group_by_folder_plain_file_path_is_not_a_group_key() -> None:
     assert "top.md" not in [folder for _source, folder in groups]
 
 
+# ---------- affected_folder_keys (phase 127, task 02) ----------
+
+
+def test_affected_folder_keys_root_level_file_is_only_the_root() -> None:
+    """A root-level file (``folder_of == ""``) affects ONLY the source
+    root — there is no folder prefix to walk."""
+    assert affected_folder_keys([("S", "a.md")]) == {("S", "")}
+
+
+def test_affected_folder_keys_nested_file_walks_the_folder_chain() -> None:
+    """A doc under ``a/b/`` affects ``a/b``, ``a``, and the source
+    root — the SAME prefix walk ``group_by_folder`` pass 1 uses (one
+    concept)."""
+    assert affected_folder_keys([("S", "a/b/c.md")]) == {
+        ("S", ""),
+        ("S", "a"),
+        ("S", "a/b"),
+    }
+
+
+def test_affected_folder_keys_multi_source_set_semantics() -> None:
+    """Keys are per source (the PK is ``(source, folder_path)``), and
+    the result is a SET — a repeated chain adds no new keys, and an
+    unrelated source keeps its own root."""
+    keys = affected_folder_keys(
+        [
+            ("S", "a/b/c.md"),
+            ("T", "x.md"),
+            ("S", "a/d.md"),  # the same chain as the first entry — no new keys
+        ]
+    )
+    assert keys == {("S", ""), ("S", "a"), ("S", "a/b"), ("T", "")}
+
+
+def test_affected_folder_keys_fragment_paths_fold_into_their_parent_folder() -> None:
+    """Embedded-image fragment paths (phase 126: ``x.pdf#page=…#img=…``)
+    get NO special case: the fragment never holds a ``/``, so
+    ``folder_of`` folds them into the parent folder's chain (a
+    root-level fragment affects only the source root)."""
+    assert affected_folder_keys([("S", "dir/img.pdf#page=2#img=0")]) == {
+        ("S", ""),
+        ("S", "dir"),
+    }
+    assert affected_folder_keys([("S", "img.pdf#page=1#img=0")]) == {("S", "")}
+
+
+def test_affected_folder_keys_empty_input_is_an_empty_set() -> None:
+    assert affected_folder_keys([]) == set()
+
+
 # ---------- build_folder_summary_prompt: system ----------
 
 
@@ -1093,3 +1144,332 @@ def test_on_progress_none_is_a_zero_cost_noop(db: Session, clean_tables) -> None
         ("FSU", "a/b"): REPLY,
         ("FSU-solo", ""): REPLY,
     }
+
+
+# ---------- generate_folder_summaries(changed_docs=…) (phase 127, task 02) ----------
+
+
+def _seed_three_sources(db: Session) -> None:
+    """The restricted-mode catalogue (three sources): A is nested
+    (``a/`` and ``a/b/``), B is a sibling single-folder source, and C
+    is a single-doc root — six candidate keys total: A ×
+    {``""``, ``a``, ``a/b``}, B × {``""``, ``b``}, C × {``""``}.
+    """
+    _add_doc(db, "A", "a/b/one.md", "A One", "A one lead.")
+    _add_doc(db, "A", "a/b/two.md", "A Two")
+    _add_doc(db, "A", "a/three.md", "A Three")
+    _add_doc(db, "B", "b/one.md", "B One")
+    _add_doc(db, "C", "solo.md", "C Solo")
+
+
+def _stamps(
+    db: Session, keys: set[tuple[str, str]]
+) -> dict[tuple[str, str], object]:
+    """The stored rows' ``updated_at`` for *keys* (raw SQL — the
+    byte-identity comparisons stay honest)."""
+    return {
+        (source, folder): _updated_at(db, source, folder) for source, folder in keys
+    }
+
+
+def _freeze(
+    db: Session, values: dict[tuple[str, str], str]
+) -> dict[tuple[str, str], object]:
+    """Overwrite the stored summaries for *values*' keys with DISTINCT
+    frozen text (a "regenerated" row is then distinguishable from an
+    "untouched" one), and return the pre-run stamps."""
+    for (source, folder_path), value in values.items():
+        db.execute(
+            text(
+                "UPDATE folder_summaries SET summary = :v"
+                " WHERE source = :s AND folder_path = :f"
+            ),
+            {"v": value, "s": source, "f": folder_path},
+        )
+    db.commit()
+    return _stamps(db, set(values))
+
+
+def test_changed_docs_restricted_regeneration_only_affected_keys(
+    db: Session, clean_tables
+) -> None:
+    """One changed file in A/``a/b/`` → exactly that file's folder
+    chain (``a/b``, ``a``, and the A root) regenerates: one lite call
+    per affected non-manual key, in sorted order, fresh rows. Every
+    OTHER stored row — same-source siblings and other sources — keeps
+    its summary text AND ``updated_at`` byte-identical (zero lite
+    calls for them)."""
+    _seed_three_sources(db)
+    asyncio.run(generate_folder_summaries(db, _FakeLLM()))
+    db.commit()
+    frozen = {
+        ("A", ""): "A root frozen",
+        ("A", "a"): "A/a frozen",
+        ("A", "a/b"): "A/a/b frozen",
+        ("B", ""): "B root frozen",
+        ("B", "b"): "B/b frozen",
+        ("C", ""): "C root frozen",
+    }
+    stamps = _freeze(db, frozen)
+
+    llm = _FakeLLM()
+    stats = asyncio.run(
+        generate_folder_summaries(db, llm, changed_docs=[("A", "a/b/one.md")])
+    )
+    assert stats == {"generated": 3, "failed": 0, "pruned": 0, "kept_manual": 0}
+    assert llm.calls == 3, "one lite call per AFFECTED non-manual key — nothing else"
+    assert [user.splitlines()[0] for _s, user in llm.requests] == [
+        "Folder: A",
+        "Folder: A/a",
+        "Folder: A/a/b",
+    ], "the affected chain in sorted (source, folder_path) order"
+
+    stored = _rows(db)
+    assert stored[("A", "")] == REPLY
+    assert stored[("A", "a")] == REPLY
+    assert stored[("A", "a/b")] == REPLY, "the changed file's own folder regenerates"
+    assert stored[("B", "")] == "B root frozen"
+    assert stored[("B", "b")] == "B/b frozen"
+    assert stored[("C", "")] == "C root frozen", (
+        "the unaffected rows keep their stored text"
+    )
+    for key in (("B", ""), ("B", "b"), ("C", "")):
+        assert _updated_at(db, *key) == stamps[key], (
+            "the unaffected rows are never re-stamped"
+        )
+
+
+def test_changed_docs_manual_affected_row_is_skipped(db: Session, clean_tables) -> None:
+    """A manual row INSIDE the affected chain is skipped exactly as in
+    full mode (phase 97): ``kept_manual``, no lite call, text AND
+    ``updated_at`` byte-identical — the restriction changes the key
+    set, not the manual-row rule."""
+    _seed_three_sources(db)
+    asyncio.run(generate_folder_summaries(db, _FakeLLM()))
+    db.commit()
+    manual_text = "Owner's words about A/a."
+    row = db.get(FolderSummary, ("A", "a"))
+    assert row is not None
+    row.summary = manual_text
+    row.manually_edited = True
+    db.commit()
+    stamp_before = _updated_at(db, "A", "a")
+
+    llm = _FakeLLM()
+    stats = asyncio.run(
+        generate_folder_summaries(db, llm, changed_docs=[("A", "a/b/one.md")])
+    )
+    assert stats == {"generated": 2, "failed": 0, "pruned": 0, "kept_manual": 1}
+    assert llm.calls == 2, "the manual key burns zero lite calls"
+    assert [user.splitlines()[0] for _s, user in llm.requests] == [
+        "Folder: A",
+        "Folder: A/a/b",
+    ]
+    stored = _rows(db)
+    assert stored[("A", "a")] == manual_text, "the owner's text is never rewritten"
+    assert _updated_at(db, "A", "a") == stamp_before, ("never re-stamped")
+    assert stored[("A", "")] == REPLY and stored[("A", "a/b")] == REPLY
+    assert stored[("B", "")] == REPLY and stored[("B", "b")] == REPLY
+    assert stored[("C", "")] == REPLY, "the unaffected rows are left as they were"
+
+
+def test_changed_docs_failing_affected_folder_fails_soft(db: Session, clean_tables) -> None:
+    """Per-folder fail-soft under the restriction: the failing
+    affected key is attempted and counted, its PREVIOUS row is kept
+    (an old summary is better than none), the other affected keys
+    still land, and the unaffected rows are untouched."""
+    _seed_three_sources(db)
+    asyncio.run(generate_folder_summaries(db, _FakeLLM()))
+    db.commit()
+    # A DISTINCT previous text on the failing folder, so "kept" is
+    # visible (it would otherwise be the REPLY either way).
+    db.execute(
+        text(
+            "UPDATE folder_summaries SET summary = 'old A/a/b'"
+            " WHERE (source, folder_path) = ('A', 'a/b')"
+        )
+    )
+    db.commit()
+    stamps = _stamps(db, {("B", ""), ("B", "b"), ("C", "")})
+
+    llm = _FakeLLM(fail_folders=("A/a/b",))
+    stats = asyncio.run(
+        generate_folder_summaries(db, llm, changed_docs=[("A", "a/b/one.md")])
+    )
+    assert stats == {"generated": 2, "failed": 1, "pruned": 0, "kept_manual": 0}
+    assert llm.calls == 3, "the failing key is attempted (the hook's premise)"
+    stored = _rows(db)
+    assert stored[("A", "a/b")] == "old A/a/b", "the previous row survives"
+    assert stored[("A", "")] == REPLY and stored[("A", "a")] == REPLY
+    for key, stamp in stamps.items():
+        assert _updated_at(db, *key) == stamp, "the unaffected rows are untouched"
+
+
+def test_changed_docs_multi_source_change_set(db: Session, clean_tables) -> None:
+    """A change set spanning SOURCES: the affected keys are the UNION
+    of each document's chain (A's nested chain + B's root and ``b``),
+    sorted across sources; the unaffected C root row keeps its text
+    AND ``updated_at``."""
+    _seed_three_sources(db)
+    asyncio.run(generate_folder_summaries(db, _FakeLLM()))
+    db.commit()
+    c_stamp = _updated_at(db, "C", "")
+
+    llm = _FakeLLM()
+    stats = asyncio.run(
+        generate_folder_summaries(
+            db, llm, changed_docs=[("A", "a/b/two.md"), ("B", "b/one.md")]
+        )
+    )
+    assert stats == {"generated": 5, "failed": 0, "pruned": 0, "kept_manual": 0}
+    assert llm.calls == 5
+    assert [user.splitlines()[0] for _s, user in llm.requests] == [
+        "Folder: A",
+        "Folder: A/a",
+        "Folder: A/a/b",
+        "Folder: B",
+        "Folder: B/b",
+    ]
+    stored = _rows(db)
+    assert stored[("C", "")] == REPLY
+    assert _updated_at(db, "C", "") == c_stamp
+
+
+def test_changed_docs_gap_in_affected_folder_is_generated(
+    db: Session, clean_tables
+) -> None:
+    """An affected folder with NO stored row (a gap in the chain) is a
+    candidate and gets generated like any other affected key — while a
+    gap in an UNAFFECTED folder stays the unchanged-walk gap probe's
+    job (phase 96): the restriction never widens beyond the affected
+    set."""
+    _seed_three_sources(db)
+    asyncio.run(generate_folder_summaries(db, _FakeLLM()))
+    db.commit()
+    db.execute(
+        text(
+            "DELETE FROM folder_summaries"
+            " WHERE (source, folder_path) IN (('A', 'a/b'), ('B', ''))"
+        )
+    )
+    db.commit()
+
+    llm = _FakeLLM()
+    stats = asyncio.run(
+        generate_folder_summaries(db, llm, changed_docs=[("A", "a/b/one.md")])
+    )
+    assert stats == {"generated": 3, "failed": 0, "pruned": 0, "kept_manual": 0}
+    assert llm.calls == 3
+    stored = _rows(db)
+    assert stored[("A", "a/b")] == REPLY, "the gap in the affected chain is filled"
+    assert ("B", "") not in stored, (
+        "the gap in the UNAFFECTED source is left for the gap probe"
+    )
+    assert stored[("B", "b")] == REPLY
+
+
+def test_changed_docs_vanished_affected_folder_is_pruned_not_generated(
+    db: Session, clean_tables
+) -> None:
+    """The ASSUMPTION pinned: an affected folder that no longer holds
+    any document is a PRUNE-pass job, not a generation's — the pruned
+    document's chain (root + its now-vanished folder) is in the change
+    set, only the surviving candidate (the root) generates, the
+    vanished folder's row is dropped, and the unaffected rows keep
+    their ``updated_at``."""
+    _add_doc(db, "A", "a/b/one.md", "A One")
+    _add_doc(db, "A", "a/b/two.md", "A Two")
+    _add_doc(db, "A", "gone/old.md", "Gone")
+    asyncio.run(generate_folder_summaries(db, _FakeLLM()))
+    db.commit()
+    stamps = _stamps(db, {("A", "a"), ("A", "a/b")})
+
+    db.execute(text("DELETE FROM documents WHERE path = 'gone/old.md'"))
+    db.commit()
+
+    llm = _FakeLLM()
+    stats = asyncio.run(
+        generate_folder_summaries(db, llm, changed_docs=[("A", "gone/old.md")])
+    )
+    # affected = {("A", ""), ("A", "gone")} — the only surviving
+    # candidate in that set is the source root.
+    assert stats == {"generated": 1, "failed": 0, "pruned": 1, "kept_manual": 0}
+    assert llm.calls == 1
+    assert [user.splitlines()[0] for _s, user in llm.requests] == ["Folder: A"]
+    stored = _rows(db)
+    assert ("A", "gone") not in stored, "the vanished folder's row is pruned"
+    assert stored[("A", "")] == REPLY, "the root's subtree changed — regenerated"
+    assert stored[("A", "a")] == REPLY and stored[("A", "a/b")] == REPLY
+    for key, stamp in stamps.items():
+        assert _updated_at(db, *key) == stamp, "the unaffected rows are untouched"
+
+
+def test_changed_docs_hook_total_is_the_affected_count(
+    db: Session, clean_tables
+) -> None:
+    """The progress hook (phase 98) under the restriction fires exactly
+    ``len(affected)`` times with ``total == len(affected)`` — the sync
+    status's ``summaries_total`` then reports the affected count
+    through the existing hook — in sorted key order."""
+    _seed_three_sources(db)
+    asyncio.run(generate_folder_summaries(db, _FakeLLM()))
+    db.commit()
+    llm = _FakeLLM()
+    events, record = _record_progress()
+    stats = asyncio.run(
+        generate_folder_summaries(
+            db, llm, changed_docs=[("A", "a/b/one.md")], on_progress=record
+        )
+    )
+    assert events == [
+        (1, 3, "A", ""),
+        (2, 3, "A", "a"),
+        (3, 3, "A", "a/b"),
+    ]
+    assert llm.calls == 3
+    assert stats["generated"] == 3
+
+
+def test_changed_docs_none_is_the_pre_127_full_regeneration(
+    db: Session, clean_tables
+) -> None:
+    """The default (no ``changed_docs`` kwarg) is the pre-127 FULL
+    regeneration over the same three-source catalogue: every
+    candidate key, one call each, the 4-field stats shape unchanged —
+    the ``scripts/import_docs.py`` CLI contract."""
+    _seed_three_sources(db)
+    llm = _FakeLLM()
+    stats = asyncio.run(generate_folder_summaries(db, llm))
+    assert stats == {"generated": 6, "failed": 0, "pruned": 0, "kept_manual": 0}
+    assert llm.calls == 6, "no kwarg → every candidate burns one call"
+    assert set(_rows(db)) == {
+        ("A", ""), ("A", "a"), ("A", "a/b"),
+        ("B", ""), ("B", "b"),
+        ("C", ""),
+    }
+
+
+def test_changed_docs_empty_set_generates_nothing_but_still_prunes(
+    db: Session, clean_tables
+) -> None:
+    """``changed_docs=[]`` (a change set that IS given but empty)
+    restricts the generation to zero keys — no lite calls, no
+    re-stamps, the hook never fires — while the prune pass still runs
+    in the given mode (the sweep is mode-independent, phase 97)."""
+    _seed_three_sources(db)
+    asyncio.run(generate_folder_summaries(db, _FakeLLM()))
+    db.commit()
+    db.add(FolderSummary(source="A", folder_path="gone/old", summary="stale"))
+    db.commit()
+    stamps = _stamps(db, {("A", ""), ("A", "a"), ("A", "a/b"), ("B", ""), ("C", "")})
+
+    llm = _FakeLLM()
+    events, record = _record_progress()
+    stats = asyncio.run(
+        generate_folder_summaries(db, llm, changed_docs=[], on_progress=record)
+    )
+    assert stats == {"generated": 0, "failed": 0, "pruned": 1, "kept_manual": 0}
+    assert llm.calls == 0
+    assert events == [], "zero keys → the hook never fires"
+    for key, stamp in stamps.items():
+        assert _updated_at(db, *key) == stamp, "no row is re-stamped"

tests/unit/test_importer.py

@@ -15,9 +15,11 @@ from __future__ import annotations
 
 import asyncio
 import logging
+import os
 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
@@ -28,6 +30,7 @@ from app.models import Chunk, Document
 from app.rag.importer import (
     EXCLUDED_DIRS,
     ImportSummary,
+    _prune,
     _store_summary,
     import_sources,
     iter_importable_files,
@@ -1194,3 +1197,236 @@ def test_progress_hook_with_limit_keeps_full_total(db, tmp_path: Path) -> None:
         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)

tests/unit/test_sync_button.py

@@ -40,7 +40,7 @@ from __future__ import annotations
 import asyncio
 import re
 import threading
-from collections.abc import Callable, Iterator
+from collections.abc import Callable, Iterator, Sequence
 from datetime import datetime
 from pathlib import Path
 
@@ -819,12 +819,15 @@ def _patch_sync_seams(
         *,
         skip: bool = False,
         only_missing: bool = False,
+        changed_docs: Sequence[tuple[str, str]] | None = None,  # phase 127
         on_progress: Callable[[int, int, str, str], None] | None = None,
     ) -> dict[str, int]:
         # Phase 98 (task 01): the runner wires the progress hook into
         # the generation branches — the fake accepts it (the state-
         # machine tests' canned run never fires it: zero summary
-        # counters, like a no-gap skip).
+        # counters, like a no-gap skip). Phase 127 (task 03): the
+        # changed-KB branch passes the change set as changed_docs= —
+        # the fake accepts it (these canned runs record nothing).
         return {"generated": 0, "failed": 0, "pruned": 0}
 
     monkeypatch.setattr(sync_api, "generate_folder_summaries", fake_folder_summaries)