"""Unit: folder summary storage + generator (phase 94, task 01).

The prompt/grouping tests are pure (no DB): ``FOLDER_SUMMARY_MODE``
system prompt, the ``folder_of`` / ``group_by_folder`` recursive-subtree
concept, and the user-message cap with the shared ``[…truncated…]``
marker. The generator tests run against the local compose Postgres
(preferred — real upsert/prune on the ``folder_summaries`` table),
skipping with clear instructions when the stack is not up — same
pattern as ``test_overview.py``.
"""
from __future__ import annotations

import asyncio
import logging
import uuid
from datetime import UTC, datetime
from typing import Any

import pytest
from sqlalchemy import text
from sqlalchemy.orm import Session

from app.config import Settings, get_settings
from app.db import SessionLocal
from app.models import Document, FolderSummary
from app.rag.folder_summaries import (
    FOLDER_HEADER_PREFIX,
    FOLDER_SUMMARY_INSTRUCTION,
    FOLDER_SUMMARY_MODE,
    MIN_DOCS_PER_FOLDER,
    SYSTEM_PROMPT,
    affected_folder_keys,
    build_folder_summary_prompt,
    folder_of,
    generate_folder_summaries,
    group_by_folder,
    missing_folder_summaries,
    summarize_folder,
)
from app.rag.llm import LLMError
from app.rag.retriever import TRUNCATION_MARKER
from tests.e2e.mock_llm import compose_answer

REPLY = "Covers lab automation runbooks: inventories, playbooks, and schedules."


class _FakeLLM:
    """Duck-typed stand-in for ``LLMClient`` (``chat`` + ``settings``).

    Records each ``(system, user)`` request and the ``model`` kwarg;
    returns the canned reply, or raises — either a fixed exception or a
    per-folder failure keyed on the user message's FIRST LINE (the
    ``Folder: …`` header) — EXACT match, so a nested folder's header
    (``Folder: S/a/b``) can never shadow its parent's (``Folder: S/a``)
    (the per-folder fail-soft tests).
    """

    def __init__(
        self,
        reply: str = REPLY,
        fail_folders: tuple[str, ...] = (),
        fail: Exception | None = None,
    ) -> None:
        self._reply = reply
        self._fail_folders = tuple(fail_folders)
        self._fail = fail
        self.settings = Settings(_env_file=None)  # pyright: ignore[reportCallIssue]
        self.calls = 0
        self.model: str | None = None
        self.requests: list[tuple[str, str]] = []

    async def chat(
        self, messages: list[dict[str, str]], model: str | None = None
    ) -> str:
        self.calls += 1
        self.model = model
        system = messages[0]["content"]
        user = messages[-1]["content"]
        self.requests.append((system, user))
        header = user.splitlines()[0] if user else ""
        for folder in self._fail_folders:
            if header == FOLDER_HEADER_PREFIX + folder:
                raise LLMError(f"simulated lite-model failure for {folder}")
        if self._fail is not None:
            raise self._fail
        return self._reply


# ---------- folder_of ----------


def test_folder_of_root_level_file_is_empty() -> None:
    assert folder_of("a.md") == ""


def test_folder_of_one_level() -> None:
    assert folder_of("a/b.md") == "a"


def test_folder_of_deep_path() -> None:
    assert folder_of("a/b/c/d.md") == "a/b/c"


def test_folder_of_iterating_walks_prefixes_to_root() -> None:
    """Iterating ``folder_of`` over its own result walks the folder
    prefixes nearest-first, ending at the root (the grouping walk)."""
    folder = folder_of("a/b/c.md")
    chain: list[str] = []
    while folder:
        chain.append(folder)
        folder = folder_of(folder)
    assert chain == ["a/b", "a"]  # plus the "" root the grouping adds


# ---------- group_by_folder ----------


def test_group_by_folder_root_file_lands_only_in_source_root() -> None:
    rows = [("S", "top.md", "T", None)]
    groups = group_by_folder(rows)
    assert set(groups) == {("S", "")}
    assert groups[("S", "")] == rows


def test_group_by_folder_nested_multi_source_recursive_subtree() -> None:
    """A doc under ``a/b/`` is present in the ``a``, ``a/b``, and ``""``
    groups (recursive subtree — the ``ls`` count scope, one concept);
    per source the candidates are ``""`` + every distinct folder
    prefix; group lists keep the input (catalogue) order."""
    rows = [
        ("S", "a/b/c.md", "C", None),
        ("S", "a/b/d.md", "D", None),
        ("S", "a/x.md", "X", None),
        ("S", "top.md", "T", None),
        ("T", "a/b/e.md", "E", None),
    ]
    groups = group_by_folder(rows)
    assert set(groups) == {
        ("S", ""),
        ("S", "a"),
        ("S", "a/b"),
        ("T", ""),
        ("T", "a"),
        ("T", "a/b"),
    }
    # The recursive-subtree concept: a/b/ docs in the a/, a/b/, and ""
    # groups alike — exactly the set each level's ls count shows.
    assert [r[1] for r in groups[("S", "a/b")]] == ["a/b/c.md", "a/b/d.md"]
    assert [r[1] for r in groups[("S", "a")]] == ["a/b/c.md", "a/b/d.md", "a/x.md"]
    assert [r[1] for r in groups[("S", "")]] == [
        "a/b/c.md",
        "a/b/d.md",
        "a/x.md",
        "top.md",
    ]
    # Multi-source: the same folder prefix under another source is a
    # separate group (PK is (source, folder_path)).
    assert [r[1] for r in groups[("T", "a/b")]] == ["a/b/e.md"]
    assert [r[1] for r in groups[("T", "")]] == ["a/b/e.md"]
    # Input (catalogue) order is preserved inside each group.
    assert [r[2] for r in groups[("S", "")]] == ["C", "D", "X", "T"]


def test_group_by_folder_single_doc_folder_is_a_group_too() -> None:
    """Grouping is pure subtree membership: at the
    :data:`MIN_DOCS_PER_FOLDER` = 1 rule every group (≥ 1 docs) is
    already a generator candidate, so grouping and candidacy agree
    (the generator's filter is an inert safety net — pinned by the
    generator tests, not the grouping)."""
    rows = [("S", "a/only.md", "O", None)]
    groups = group_by_folder(rows)
    assert len(groups[("S", "a")]) == 1  # present, but below the minimum


def test_group_by_folder_doc_path_equal_to_a_folder_prefix_counts_for_it() -> None:
    """The count rule's ``path == folder`` arm: a document whose path
    IS one of the source's folder prefixes (a file sharing its name
    with a directory) belongs to that folder's group too — the grouping
    stays EXACTLY the set the ``ls`` count rule counts (path equal or
    starting with ``folder + "/"``), while the returned keys remain the
    true folder prefixes only (no file-path keys)."""
    rows = [
        ("S", "a/b", "B", None),  # a file named "b" ... (its path is a folder prefix)
        ("S", "a/b/c.md", "C", None),  # ... and a real folder "a/b/" holding a doc
        ("S", "a/x.md", "X", None),
    ]
    groups = group_by_folder(rows)
    assert set(groups) == {("S", ""), ("S", "a"), ("S", "a/b")}, (
        "the keys stay the true folder prefixes — the file's own path adds no key"
    )
    assert [r[1] for r in groups[("S", "a/b")]] == ["a/b", "a/b/c.md"]
    assert [r[1] for r in groups[("S", "a")]] == ["a/b", "a/b/c.md", "a/x.md"]
    assert [r[1] for r in groups[("S", "")]] == ["a/b", "a/b/c.md", "a/x.md"]


def test_group_by_folder_plain_file_path_is_not_a_group_key() -> None:
    """A file path that is NO folder prefix (no doc under it) adds no
    group key of its own — the ``path == folder`` arm only fires when
    the path really is a prefix of the catalogue."""
    rows = [("S", "top.md", "T", None), ("S", "a/one.md", "O", None)]
    groups = group_by_folder(rows)
    assert set(groups) == {("S", ""), ("S", "a")}
    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 ----------


def test_system_prompt_has_marker_and_locked_instruction() -> None:
    assert SYSTEM_PROMPT.startswith(FOLDER_SUMMARY_MODE)
    assert FOLDER_SUMMARY_INSTRUCTION in SYSTEM_PROMPT
    for fragment in (
        "1-3 sentence",
        "plain-text summary",
        "natural language",
        "Do not use markdown",
        "not in the list",
    ):
        assert fragment in SYSTEM_PROMPT
    system, _ = build_folder_summary_prompt("S", "a/b", [])
    assert system == SYSTEM_PROMPT
    assert FOLDER_SUMMARY_MODE in system  # the marker the E2E mock keys on


def test_instruction_pins_the_variety_nudge_and_preserved_constraints() -> None:
    """Phase 155 task 01: the locked instruction keeps the four phase-94
    constraints (1–3 sentences; plain text; no markdown; no invention)
    and ADDS the variety direction — the two forbidden openers are the
    pin (an explicit NOT, not a mere mention)."""
    # The four preserved constraints (phase 94 — unchanged).
    for fragment in (
        "1-3 sentence",
        "plain-text",
        "natural language",
        "Do not use markdown",
        "Do not invent anything that is not in the list",
    ):
        assert fragment in FOLDER_SUMMARY_INSTRUCTION, fragment
    # The variety direction (phase 155): the "varied, natural way"
    # direction must survive the grammar polish.
    assert "Open in a varied, natural way" in FOLDER_SUMMARY_INSTRUCTION
    assert "state what the folder is for" in FOLDER_SUMMARY_INSTRUCTION
    # The two forbidden openers, banned by an explicit NOT.
    assert "Do NOT begin with" in FOLDER_SUMMARY_INSTRUCTION
    assert "'This folder contains'" in FOLDER_SUMMARY_INSTRUCTION
    assert "'The <folder name> folder contains'" in FOLDER_SUMMARY_INSTRUCTION


def test_marker_and_header_prefix_stay_byte_identical() -> None:
    """Phase 155 task 01: the E2E mock keys on ``FOLDER_SUMMARY_MODE``
    and the ``FOLDER_HEADER_PREFIX`` line — the instruction nudge must
    not touch them (the mock-contract pin).``SYSTEM_PROMPT`` keeps the
    marker-first shape: the nudge lands AFTER the marker."""
    assert FOLDER_SUMMARY_MODE == "FOLDER_SUMMARY_MODE"
    assert FOLDER_HEADER_PREFIX == "Folder: "
    assert SYSTEM_PROMPT.startswith("FOLDER_SUMMARY_MODE: ")


# ---------- build_folder_summary_prompt: user ----------


def test_user_prompt_header_names_the_folder() -> None:
    """The first line is the ``FOLDER_HEADER_PREFIX`` header the E2E
    mock parses: ``<source>`` for the root, ``<source>/<folder_path>``
    for a folder."""
    _, user = build_folder_summary_prompt("Homelab", "deployments/ansible", [])
    assert user == FOLDER_HEADER_PREFIX + "Homelab/deployments/ansible"
    _, user = build_folder_summary_prompt("Homelab", "", [])
    assert user == FOLDER_HEADER_PREFIX + "Homelab"


def test_user_lines_carry_path_title_and_first_summary_line() -> None:
    docs = [
        ("S", "a/b/one.md", "One", "First lead.\nSecond line.\nSource: S/a/b/one.md"),
        ("S", "a/b/two.md", "Two", None),
    ]
    system, user = build_folder_summary_prompt("S", "a/b", docs)
    assert system == SYSTEM_PROMPT
    assert user == (
        "Folder: S/a/b\n"
        "a/b/one.md — One — First lead.\n"
        "a/b/two.md — Two"
    )


def test_user_line_omits_summary_field_when_absent_or_blank() -> None:
    docs = [
        ("S", "a/x.md", "X", None),
        ("S", "a/y.md", "Y", "  \n\t "),
    ]
    _, user = build_folder_summary_prompt("S", "a", docs)
    assert user == "Folder: S/a\na/x.md — X\na/y.md — Y"
    assert " — " in user  # the path — title join only
    assert not any(line.endswith(" — ") for line in user.splitlines())


def test_user_line_uses_only_first_summary_line() -> None:
    docs = [
        ("S", "a/x.md", "X", "First line.\nSecond line.\nSource: S/a/x.md"),
    ]
    _, user = build_folder_summary_prompt("S", "a", docs)
    assert user == "Folder: S/a\na/x.md — X — First line."
    assert "Second line" not in user
    assert "Source:" not in user


def test_user_prompt_truncated_with_marker_when_over_custom_cap() -> None:
    docs = [("S", f"a/f{i}.md", f"T{i}", None) for i in range(10)]
    _, full = build_folder_summary_prompt("S", "a", docs, max_chars=10_000)
    cap = 30
    _, user = build_folder_summary_prompt("S", "a", docs, max_chars=cap)
    assert user == full[:cap] + "\n" + TRUNCATION_MARKER
    assert user.endswith(TRUNCATION_MARKER)
    assert len(user) > cap  # the marker makes the cut visible past the cap


def test_user_prompt_at_exact_cap_not_truncated() -> None:
    docs = [("S", "a/x.md", "X", None)]  # "Folder: S/a\na/x.md — X" = 22 chars
    _, user = build_folder_summary_prompt("S", "a", docs, max_chars=22)
    assert user == "Folder: S/a\na/x.md — X"
    assert TRUNCATION_MARKER not in user


def test_user_prompt_truncated_at_default_cap() -> None:
    """No explicit cap → ``BOR_FOLDER_SUMMARY_INPUT_MAX_CHARS`` (read
    from the live settings, so the test holds for any configured
    value)."""
    cap = get_settings().folder_summary_input_max_chars
    docs = [("S", f"a/f{i}.md", "T", None) for i in range(3_000)]
    _, user = build_folder_summary_prompt("S", "a", docs)
    assert user.endswith(TRUNCATION_MARKER)
    body = user.removesuffix("\n" + TRUNCATION_MARKER)
    assert len(body) == cap  # cut exactly at the cap, marker on its own line
    assert "f2999.md" not in body  # the overflow never reaches the model


# ---------- summarize_folder ----------


# ---------- the E2E mock's FOLDER_SUMMARY_MODE branch ----------


def _mock_body(system: str, user: str) -> dict[str, Any]:
    """A minimal chat-completion body for the mock's ``compose_answer``."""
    return {"messages": [
        {"role": "system", "content": system},
        {"role": "user", "content": user},
    ]}


def test_mock_returns_canned_folder_summary_naming_the_folder() -> None:
    """The deterministic E2E mock keys on the ``FOLDER_SUMMARY_MODE``
    marker in the system prompt and returns the canned one-liner naming
    the folder from the ``Folder: …`` header — driven through the
    GENERATOR's real prompt, so the two can never drift (the drill-down
    E2E asserts on this exact template). Phase 155 task 01: the canned
    voice is the natural-opener ``Documentation for <folder>.``."""
    system, user = build_folder_summary_prompt(
        "Homelab", "deployments/ansible",
        [("Homelab", "deployments/ansible/lab-inventory.md", "Lab Inventory", None)],
    )
    assert compose_answer(_mock_body(system, user)) == (
        "Documentation for Homelab/deployments/ansible."
    )
    # The source-root row names the source itself.
    system, user = build_folder_summary_prompt("Homelab", "",
                                               [("Homelab", "top.md", "Top", None)])
    assert compose_answer(_mock_body(system, user)) == (
        "Documentation for Homelab."
    )


def test_mock_canned_folder_reply_shape_and_determinism() -> None:
    """Phase 155 task 01: the canned folder reply is the new voice —
    it NAMES the folder (from the ``Folder: …`` header), AVOIDS the
    two forbidden openers, and is a PURE function of the request (the
    determinism contract: two identical requests → identical replies).
    """
    def reply_for(source: str, folder_path: str) -> str:
        system, user = build_folder_summary_prompt(
            source, folder_path, [(source, "x.md", "X", None)]
        )
        return compose_answer(_mock_body(system, user))

    first = reply_for("Homelab", "deployments/ansible")
    second = reply_for("Homelab", "deployments/ansible")  # same request → same reply
    assert first == second, "pure function of the request (determinism contract)"
    assert "Homelab/deployments/ansible" in first, "the reply names the folder"
    # The two forbidden openers (phase 155's pin), case-insensitively.
    assert not first.lower().startswith("this folder contains"), (
        "the formulaic opener must not survive the canned reply"
    )
    assert not first.lower().startswith(
        "the homelab/deployments/ansible folder contains"
    ), "the named-folder formulaic opener must not survive either"
    # A different folder yields a different (but same-shape) reply —
    # still names its own folder, still neither forbidden opener.
    root = reply_for("Homelab", "")
    assert root != first, "the reply is a function of the request, not a constant"
    assert "Homelab" in root
    assert not root.lower().startswith("this folder contains")
    assert not root.lower().startswith("the homelab folder contains")


def test_mock_folder_marker_is_not_shadowed_by_the_summary_branch() -> None:
    """``FOLDER_SUMMARY_MODE`` contains ``SUMMARY_MODE`` as a substring —
    the mock must check the folder branch FIRST, or every folder call
    would land in the document-summary digest (regression pin)."""
    system, user = build_folder_summary_prompt(
        "S", "a", [("S", "a/x.md", "X", None)]
    )
    assert "SUMMARY_MODE" in system  # the shadowing hazard is real
    answer = compose_answer(_mock_body(system, user))
    assert answer == "Documentation for S/a."
    assert not answer.startswith("This document covers")


def test_summarize_folder_happy_path_returns_trimmed_text() -> None:
    docs = [("S", "a/x.md", "X", None)]
    llm = _FakeLLM(reply=f"  {REPLY}  \n")
    out = asyncio.run(summarize_folder("S", "a", docs, llm))
    assert out == REPLY  # the model's text, trimmed
    assert llm.calls == 1


def test_summarize_folder_calls_the_configured_summary_model_with_marker() -> None:
    docs = [("S", "a/x.md", "X", "X lead.")]
    llm = _FakeLLM()
    asyncio.run(summarize_folder("S", "a", docs, llm))
    assert llm.model == llm.settings.llm_summary_model  # the ``lite`` default
    assert llm.model == "lite"
    system, user = llm.requests[0]
    assert FOLDER_SUMMARY_MODE in system
    assert user.startswith(FOLDER_HEADER_PREFIX + "S/a")
    assert "a/x.md — X — X lead." in user


def test_summarize_folder_empty_reply_raises_llm_error() -> None:
    docs = [("S", "a/x.md", "X", None)]
    for reply in ("", "   \n\t "):
        llm = _FakeLLM(reply=reply)
        with pytest.raises(LLMError, match="empty content for S/a"):
            asyncio.run(summarize_folder("S", "a", docs, llm))


def test_summarize_folder_error_propagates() -> None:
    docs = [("S", "a/x.md", "X", None)]
    llm = _FakeLLM(fail=LLMError("simulated transport failure"))
    with pytest.raises(LLMError, match="simulated transport failure"):
        asyncio.run(summarize_folder("S", "a", docs, llm))


# ---------- generate_folder_summaries (real Postgres) ----------


def _add_doc(
    db: Session, source: str, path: str, title: str, summary: str | None = None
) -> Document:
    doc = Document(
        id=uuid.uuid4(),
        source=source,
        path=path,
        full_path=f"/tmp/{path}",
        title=title,
        content="body",
        content_hash="0" * 64,
        summary=summary,
    )
    db.add(doc)
    db.commit()
    return doc


def _truncate(db: Session) -> None:
    db.execute(text("TRUNCATE chunks, documents"))
    db.execute(text("DELETE FROM folder_summaries"))
    db.commit()


def _rows(db: Session) -> dict[tuple[str, str], str]:
    """The stored folder summaries: ``{(source, folder_path): summary}``."""
    result = db.execute(
        text("SELECT source, folder_path, summary FROM folder_summaries")
    ).all()
    return {(source, folder_path): summary for source, folder_path, summary in result}


def _seed_catalogue(db: Session) -> None:
    """The shared catalogue: FSU has four docs in three candidate
    folders (root 4, a 3, a/b 2); FSU-solo has one doc (its root
    folder is a candidate too — the ≥ 1 rule: a single-file source
    root gets a row)."""
    _add_doc(db, "FSU", "a/b/one.md", "One", "One lead.\nSource: FSU/a/b/one.md")
    _add_doc(db, "FSU", "a/b/two.md", "Two")
    _add_doc(db, "FSU", "a/three.md", "Three")
    _add_doc(db, "FSU", "root.md", "Root")
    _add_doc(db, "FSU-solo", "solo.md", "Solo")


@pytest.fixture()
def clean_tables(db: Session):
    _truncate(db)
    yield
    _truncate(db)


def test_generate_happy_path_upserts_every_candidate_folder(
    db: Session, clean_tables, caplog: pytest.LogCaptureFixture
) -> None:
    """Every folder with ≥ 1 recursive doc gets a row (the source root
    row included — ``folder_path = ''`` — and the single-doc FSU-solo
    root: the ≥ 1 rule); rows are stamped fresh; the stats dict and
    the log line are right; folders are processed in deterministic
    (source, folder_path) order."""
    _seed_catalogue(db)
    llm = _FakeLLM()
    with caplog.at_level(logging.INFO, logger="app.rag.folder_summaries"):
        stats = asyncio.run(generate_folder_summaries(db, llm))
    assert stats == {"generated": 4, "failed": 0, "pruned": 0, "kept_manual": 0}
    assert llm.calls == 4, "one lite call per candidate folder (the solo root: too)"

    stored = _rows(db)
    assert set(stored) == {("FSU", ""), ("FSU", "a"), ("FSU", "a/b"), ("FSU-solo", "")}
    assert all(summary == REPLY for summary in stored.values())
    assert stored[("FSU-solo", "")] == REPLY, (
        "the single-doc source root is a candidate at the ≥ 1 rule"
    )

    row = db.get(FolderSummary, ("FSU", "a/b"))
    assert row is not None
    assert row.summary == REPLY
    assert row.updated_at is not None
    age = datetime.now(UTC) - row.updated_at
    assert age.total_seconds() < 300, "updated_at must be a fresh UTC timestamp"

    # Deterministic (source, folder_path) order — root before the
    # nested folders, one header per call.
    assert [user.splitlines()[0] for _s, user in llm.requests] == [
        "Folder: FSU",
        "Folder: FSU/a",
        "Folder: FSU/a/b",
        "Folder: FSU-solo",
    ]
    # The recursive-subtree input: the a/ prompt carries a/b's docs too.
    a_prompt = llm.requests[1][1]
    assert "a/b/one.md — One — One lead." in a_prompt
    assert "a/three.md — Three" in a_prompt
    assert "root.md — Root" not in a_prompt

    assert (
        "folder_summaries: generated=4 failed=0 pruned=0 kept_manual=0"
        in caplog.text
    ), "the stats line must be greppable (PLAN §9 ample logging)"


def test_generate_per_folder_fail_soft_keeps_previous_and_lands_others(
    db: Session, clean_tables, caplog: pytest.LogCaptureFixture
) -> None:
    """One folder's lite failure is logged and counted, its PREVIOUS
    row is kept (an old summary is better than none), and the remaining
    folders still land — a lite outage never fails the sync."""
    _seed_catalogue(db)
    db.add(FolderSummary(source="FSU", folder_path="a/b", summary="old summary"))
    db.commit()
    llm = _FakeLLM(fail_folders=("FSU/a/b",))
    with caplog.at_level(logging.ERROR, logger="app.rag.folder_summaries"):
        stats = asyncio.run(generate_folder_summaries(db, llm))
    assert stats == {"generated": 3, "failed": 1, "pruned": 0, "kept_manual": 0}
    assert llm.calls == 4  # the failing folder was attempted too

    stored = _rows(db)
    assert stored[("FSU", "a/b")] == "old summary", (
        "the previous row survives the per-folder failure"
    )
    assert (
        stored[("FSU", "")] == REPLY
        and stored[("FSU", "a")] == REPLY
        and stored[("FSU-solo", "")] == REPLY
    ), "the other folders still land"
    assert "folder summary failed for FSU/a/b" in caplog.text
    assert "simulated lite-model failure for FSU/a/b" in caplog.text


def test_generate_per_folder_fail_soft_without_previous_row_creates_nothing(
    db: Session, clean_tables
) -> None:
    _seed_catalogue(db)
    llm = _FakeLLM(fail_folders=("FSU/a/b",))
    stats = asyncio.run(generate_folder_summaries(db, llm))
    assert stats["failed"] == 1
    stored = _rows(db)
    assert ("FSU", "a/b") not in stored, "no row must be invented for a failed folder"
    assert ("FSU", "") in stored and ("FSU", "a") in stored


def test_generate_prunes_stale_rows_and_keeps_live_ones(db: Session, clean_tables) -> None:
    """Rows for folders that VANISHED (0 recursive docs) are deleted
    (pruned/renamed — the summary would go stale); rows for folders
    that still hold ≥ 1 doc persist (an unchanged folder's summary is
    still true — regenerated in place), the 1-doc FSU-solo root
    included (it is a candidate at the ≥ 1 rule)."""
    _seed_catalogue(db)
    # A stale row for a folder no longer in the catalogue (renamed
    # away — 0 docs) + live rows with old content.
    db.add(FolderSummary(source="FSU", folder_path="gone/old", summary="stale"))
    db.add(FolderSummary(source="FSU", folder_path="a", summary="old a summary"))
    db.add(FolderSummary(source="FSU-solo", folder_path="", summary="solo stale"))
    db.commit()
    stats = asyncio.run(generate_folder_summaries(db, _FakeLLM()))
    assert stats["pruned"] == 1  # gone/old only (its folder vanished)

    stored = _rows(db)
    assert ("FSU", "gone/old") not in stored, "the stale folder row must be pruned"
    assert stored[("FSU-solo", "")] == REPLY, (
        "the 1-doc root qualifies — regenerated, not pruned"
    )
    assert ("FSU", "a") in stored, "the still-qualifying folder keeps its row"
    assert stored[("FSU", "a")] == REPLY  # regenerated, not stale
    assert stored[("FSU", "")] == REPLY and stored[("FSU", "a/b")] == REPLY


def test_generate_skip_is_a_full_noop(db: Session, clean_tables) -> None:
    """``skip=True`` (the ``--limit`` debug run): the LLM is never
    called, no rows are touched, zero stats."""
    _seed_catalogue(db)
    db.add(FolderSummary(source="FSU", folder_path="", summary="existing"))
    db.commit()
    llm = _FakeLLM()
    stats = asyncio.run(generate_folder_summaries(db, llm, skip=True))
    assert stats == {"generated": 0, "failed": 0, "pruned": 0, "kept_manual": 0}
    assert llm.calls == 0
    assert _rows(db) == {("FSU", ""): "existing"}


def test_generate_empty_kb_prunes_every_row(db: Session, clean_tables) -> None:
    """No documents → no candidate folders → every stored row is
    pruned, with zero wasted lite calls."""
    db.add(FolderSummary(source="FSU", folder_path="", summary="old"))
    db.add(FolderSummary(source="FSU", folder_path="a/b", summary="old"))
    db.commit()
    llm = _FakeLLM()
    stats = asyncio.run(generate_folder_summaries(db, llm))
    assert stats == {"generated": 0, "failed": 0, "pruned": 2, "kept_manual": 0}
    assert llm.calls == 0
    assert _rows(db) == {}


def test_generate_summarizes_folder_whose_prefix_is_also_a_doc_path(
    db: Session, clean_tables
) -> None:
    """The ``path == folder`` arm end to end: a file sharing its name
    with a directory counts toward the folder's recursive count (2
    docs → the folder is summarized, and BOTH docs are in its prompt).
    """
    _add_doc(db, "FSU", "a/b", "B")  # a file named "b" (its path is a prefix)
    _add_doc(db, "FSU", "a/b/c.md", "C")  # and a real folder "a/b/"
    llm = _FakeLLM()
    stats = asyncio.run(generate_folder_summaries(db, llm))
    assert stats["generated"] == 3  # root (2), a (2), a/b (2) — all ≥ the minimum
    stored = _rows(db)
    assert set(stored) == {("FSU", ""), ("FSU", "a"), ("FSU", "a/b")}
    a_b_prompt = [
        user for _system, user in llm.requests if user.startswith("Folder: FSU/a/b\n")
    ][0]
    assert "a/b — B" in a_b_prompt
    assert "a/b/c.md — C" in a_b_prompt


def test_generate_only_flushes_caller_commits(db: Session, clean_tables) -> None:
    """The generator only flushes — the sync path owns the transaction
    (the phase-53 ``bump_sources_version`` convention): the catalogue
    is committed (the real sync path commits the import before the
    summary hooks run), but a second session sees the generator's rows
    as NOTHING until the CALLER commits — and sees them after."""
    _add_doc(db, "FSU", "x/y/one.md", "One")
    _add_doc(db, "FSU", "x/y/two.md", "Two")
    stats = asyncio.run(generate_folder_summaries(db, _FakeLLM()))
    assert stats["generated"] == 3  # root + x + x/y — all 2 recursive docs

    with SessionLocal() as other:
        n = other.scalar(
            text("SELECT count(*) FROM folder_summaries WHERE source = 'FSU'")
        )
        assert n == 0, "unflushed-by-caller rows must not be visible yet"

    db.commit()
    with SessionLocal() as other:
        n = other.scalar(
            text("SELECT count(*) FROM folder_summaries WHERE source = 'FSU'")
        )
        assert n == 3, "the caller's commit makes the flushed rows durable"

    assert MIN_DOCS_PER_FOLDER == 1  # the ≥ 1 scope rule, pinned by name


# ---------- manually_edited (phase 97, task 01) ----------


def test_manual_row_survives_regeneration(
    db: Session, clean_tables, caplog: pytest.LogCaptureFixture
) -> None:
    """An owner-edited row is SKIPPED on regeneration (phase 97, task
    01): the fake LLM is never called for it (no ``lite`` burn on owner
    text — not even a prompt is built), its text AND ``updated_at``
    stay byte-identical, ``kept_manual`` counts it, the flag is never
    cleared, and the 4-field log line carries it (PLAN §9)."""
    _seed_catalogue(db)
    manual_text = "Owner's own words about a/."
    db.add(
        FolderSummary(
            source="FSU", folder_path="a", summary=manual_text,
            manually_edited=True,
        )
    )
    db.commit()
    stamp_before = _updated_at(db, "FSU", "a")
    assert stamp_before is not None

    llm = _FakeLLM()
    with caplog.at_level(logging.INFO, logger="app.rag.folder_summaries"):
        stats = asyncio.run(generate_folder_summaries(db, llm))

    assert stats == {"generated": 3, "failed": 0, "pruned": 0, "kept_manual": 1}
    assert llm.calls == 3, "the manual folder burns zero lite calls"
    assert [user.splitlines()[0] for _s, user in llm.requests] == [
        "Folder: FSU",
        "Folder: FSU/a/b",
        "Folder: FSU-solo",
    ], "no prompt is ever built for the owner's folder"

    stored = _rows(db)
    assert stored[("FSU", "a")] == manual_text, "the owner's text survives"
    assert _updated_at(db, "FSU", "a") == stamp_before, ("never re-stamped")
    assert (
        stored[("FSU", "")] == REPLY
        and stored[("FSU", "a/b")] == REPLY
        and stored[("FSU-solo", "")] == REPLY
    ), (
        "the non-manual candidates still regenerate (the flag is the difference)"
    )
    row = db.get(FolderSummary, ("FSU", "a"))
    assert row is not None and row.manually_edited is True, (
        "the generator never clears the flag"
    )
    assert (
        "folder_summaries: generated=3 failed=0 pruned=0 kept_manual=1"
        in caplog.text
    ), "the 4-field stats line must be greppable (PLAN §9 ample logging)"


def test_manual_row_survives_the_prune(db: Session, clean_tables) -> None:
    """A manual row is NEVER pruned and NEVER overwritten (phase 97,
    task 01): a/ and b/ each drop to 1 recursive doc — still
    candidates at the ≥ 1 rule — so the MANUAL a/ row is SKIPPED
    (owner text kept, no re-stamp, no lite burn) while the NON-manual
    b/ twin regenerates in place; the vanished folders' rows show the
    prune rule — the manual gone/manual row persists (owner content
    until cleared — the clear deletes it, so the next KB-changing sync
    regenerates an AI description) and the non-manual gone/ai twin is
    pruned; the flag is the only difference in each pair."""
    _add_doc(db, "FSU", "a/one.md", "One")
    _add_doc(db, "FSU", "a/two.md", "Two")
    _add_doc(db, "FSU", "b/one.md", "B One")
    _add_doc(db, "FSU", "b/two.md", "B Two")
    manual_text = "Owner's words about a/."
    db.add(
        FolderSummary(
            source="FSU", folder_path="a", summary=manual_text,
            manually_edited=True,
        )
    )
    db.add(FolderSummary(source="FSU", folder_path="b", summary="ai words"))
    db.add(
        FolderSummary(
            source="FSU", folder_path="gone/manual", summary="owner kept",
            manually_edited=True,
        )
    )
    db.add(FolderSummary(source="FSU", folder_path="gone/ai", summary="stale ai"))
    db.commit()

    # a/ and b/ each drop to 1 recursive doc — still candidates at
    # the ≥ 1 rule.
    db.execute(
        text(
            "DELETE FROM documents WHERE source = 'FSU'"
            " AND path IN ('a/two.md', 'b/two.md')"
        )
    )
    db.commit()

    llm = _FakeLLM()
    stats = asyncio.run(generate_folder_summaries(db, llm))

    assert stats == {"generated": 2, "failed": 0, "pruned": 1, "kept_manual": 1}
    assert llm.calls == 2, "root + the non-manual b/ regenerate; manual a/ burns nothing"
    stored = _rows(db)
    assert stored[("FSU", "a")] == manual_text, (
        "the manual row is skipped — never overwritten, even at 1 doc"
    )
    assert stored[("FSU", "b")] == REPLY, (
        "the non-manual twin regenerates — 1 doc still qualifies (the ≥ 1 rule)"
    )
    assert stored[("FSU", "gone/manual")] == "owner kept", (
        "a vanished folder's manual row is kept — owner content until cleared"
    )
    assert ("FSU", "gone/ai") not in stored, ("the non-manual twin is pruned")
    assert stored[("FSU", "")] == REPLY  # the root (2 docs) still regenerates


# ---------- missing_folder_summaries (phase 96, task 02) ----------


def test_missing_fresh_table_is_exactly_the_candidate_set(
    db: Session, clean_tables
) -> None:
    """No stored rows → every candidate folder is a gap, sorted by
    ``(source, folder_path)`` — the single-doc FSU-solo root
    included (a candidate at the ≥ 1 rule)."""
    _seed_catalogue(db)
    assert missing_folder_summaries(db) == [
        ("FSU", ""),
        ("FSU", "a"),
        ("FSU", "a/b"),
        ("FSU-solo", ""),
    ]


def test_missing_fully_populated_table_is_empty(db: Session, clean_tables) -> None:
    """Every candidate row present → no gap (the zero-burn gate case)."""
    _seed_catalogue(db)
    asyncio.run(generate_folder_summaries(db, _FakeLLM()))
    db.commit()
    assert missing_folder_summaries(db) == []


def test_missing_one_deleted_row_is_that_folder(db: Session, clean_tables) -> None:
    _seed_catalogue(db)
    asyncio.run(generate_folder_summaries(db, _FakeLLM()))
    db.commit()
    db.execute(
        text(
            "DELETE FROM folder_summaries "
            "WHERE source = 'FSU' AND folder_path = 'a'"
        )
    )
    db.commit()
    assert missing_folder_summaries(db) == [("FSU", "a")]


def test_missing_empty_kb_empty_table_is_no_gap(db: Session, clean_tables) -> None:
    """No catalogue → no candidates → ``[]`` — an empty table over an
    empty KB is not a gap (there is nothing to fill)."""
    assert missing_folder_summaries(db) == []


def test_missing_single_doc_folder_is_listed(db: Session, clean_tables) -> None:
    """The ≥ 1 rule: a single-doc folder WITHOUT a row IS a gap (it is
    a candidate — the one file line no longer exempts it), alongside
    its source root (1 doc)."""
    _add_doc(db, "FSU", "solo/one.md", "One")
    assert missing_folder_summaries(db) == [("FSU", ""), ("FSU", "solo")]


def test_missing_stale_row_is_not_a_gap(db: Session, clean_tables) -> None:
    """A stored row for a folder that dropped below the minimum
    (vanished — 0 docs) is stale, not missing — the prune pass owns
    it, the gap detector ignores it."""
    _seed_catalogue(db)
    asyncio.run(generate_folder_summaries(db, _FakeLLM()))
    db.commit()
    db.add(FolderSummary(source="FSU", folder_path="gone/old", summary="stale"))
    db.commit()
    assert missing_folder_summaries(db) == []


# ---------- generate_folder_summaries(only_missing=…) (phase 96, 02) ----------


def _updated_at(db: Session, source: str, folder_path: str) -> object:
    """The stored row's ``updated_at`` (raw SQL — bypasses the ORM
    identity map, so the before/after byte-identity comparison is
    honest)."""
    return db.execute(
        text(
            "SELECT updated_at FROM folder_summaries "
            "WHERE source = :s AND folder_path = :f"
        ),
        {"s": source, "f": folder_path},
    ).scalar_one()


def test_only_missing_fills_exactly_the_missing_keys(
    db: Session, clean_tables
) -> None:
    """Two missing + two present → exactly the missing keys are
    generated (sorted order, one lite call each); the present rows are
    byte-identical after (text AND ``updated_at``); stats right."""
    _seed_catalogue(db)
    asyncio.run(generate_folder_summaries(db, _FakeLLM()))
    db.commit()
    full = _rows(db)
    a_stamp = _updated_at(db, "FSU", "a")
    db.execute(
        text(
            "DELETE FROM folder_summaries "
            "WHERE (source, folder_path) IN (('FSU', ''), ('FSU', 'a/b'))"
        )
    )
    db.commit()
    assert missing_folder_summaries(db) == [("FSU", ""), ("FSU", "a/b")]

    llm = _FakeLLM()
    stats = asyncio.run(generate_folder_summaries(db, llm, only_missing=True))
    assert stats == {"generated": 2, "failed": 0, "pruned": 0, "kept_manual": 0}
    assert llm.calls == 2, "one call per MISSING key — zero for present rows"
    assert [user.splitlines()[0] for _s, user in llm.requests] == [
        "Folder: FSU",
        "Folder: FSU/a/b",
    ], "the missing keys in sorted (source, folder_path) order"

    assert _rows(db) == full, "the fill restores exactly the full candidate set"
    assert _updated_at(db, "FSU", "a") == a_stamp, (
        "the present row is byte-identical — never re-stamped by the fill"
    )


def test_only_missing_no_gap_burns_zero_calls(db: Session, clean_tables) -> None:
    """No gap → zero lite calls, zero rows touched, zero stats (the
    zero-burn invariant the unchanged-sync gate relies on)."""
    _seed_catalogue(db)
    asyncio.run(generate_folder_summaries(db, _FakeLLM()))
    db.commit()
    before = _rows(db)
    stamps = {f: _updated_at(db, "FSU", f) for f in ("", "a", "a/b")}
    llm = _FakeLLM()
    stats = asyncio.run(generate_folder_summaries(db, llm, only_missing=True))
    assert stats == {"generated": 0, "failed": 0, "pruned": 0, "kept_manual": 0}
    assert llm.calls == 0, "zero-burn: no gap, no lite call"
    assert _rows(db) == before
    for folder, stamp in stamps.items():
        assert _updated_at(db, "FSU", folder) == stamp, "no row re-stamped"


def test_only_missing_still_prunes_stale_rows(db: Session, clean_tables) -> None:
    """The prune pass runs in BOTH modes: the manually seeded stale row
    (folder gone from the catalogue) is pruned while the genuine
    missing folders are filled, and the present row stays untouched."""
    _seed_catalogue(db)
    db.add(FolderSummary(source="FSU", folder_path="a", summary="keep me"))
    db.add(FolderSummary(source="FSU", folder_path="gone/old", summary="stale"))
    db.commit()
    a_stamp = _updated_at(db, "FSU", "a")
    assert missing_folder_summaries(db) == [
        ("FSU", ""),
        ("FSU", "a/b"),
        ("FSU-solo", ""),
    ]

    llm = _FakeLLM()
    stats = asyncio.run(generate_folder_summaries(db, llm, only_missing=True))
    assert stats == {"generated": 3, "failed": 0, "pruned": 1, "kept_manual": 0}
    assert llm.calls == 3

    stored = _rows(db)
    assert ("FSU", "gone/old") not in stored, (
        "the stale row is pruned even under only_missing"
    )
    assert stored[("FSU", "a")] == "keep me"
    assert _updated_at(db, "FSU", "a") == a_stamp
    assert (
        stored[("FSU", "")] == REPLY
        and stored[("FSU", "a/b")] == REPLY
        and stored[("FSU-solo", "")] == REPLY
    )


def test_only_missing_fail_soft_keeps_prior_and_lands_others(
    db: Session, clean_tables
) -> None:
    """Per-folder fail-soft applies under ``only_missing`` too: the
    failing missing folder is counted and stays absent; the other
    missing folders still land; the present row is untouched."""
    _seed_catalogue(db)
    db.add(FolderSummary(source="FSU", folder_path="a", summary="keep me"))
    db.commit()
    llm = _FakeLLM(fail_folders=("FSU/a/b",))
    stats = asyncio.run(generate_folder_summaries(db, llm, only_missing=True))
    assert stats == {"generated": 2, "failed": 1, "pruned": 0, "kept_manual": 0}
    assert llm.calls == 3  # all three missing folders were attempted

    stored = _rows(db)
    assert stored[("FSU", "")] == REPLY, "the other missing folders still land"
    assert stored[("FSU-solo", "")] == REPLY, "the single-doc root still lands"
    assert ("FSU", "a/b") not in stored, "the failed folder stays absent"
    assert stored[("FSU", "a")] == "keep me", "the present row is untouched"


def test_gap_probe_subsumes_the_table_empty_gate(db: Session, clean_tables) -> None:
    """The deleted phase-94 table-empty gate probe, re-expressed through
    ``missing_folder_summaries`` (phase 96, task 03 — the probe's unit
    coverage moved here): an empty table over a populated catalogue
    means EVERY candidate is missing (the targeted fill over all
    candidates IS a full generation — the first full sync after
    migration 0017 must still generate), a populated table means no
    gap (a populated table waits for a KB change or a gap)."""
    assert missing_folder_summaries(db) == []  # the truncated table, empty KB
    _add_doc(db, "FSU", "a/one.md", "One")
    _add_doc(db, "FSU", "a/two.md", "Two")
    assert missing_folder_summaries(db) == [("FSU", ""), ("FSU", "a")]
    asyncio.run(generate_folder_summaries(db, _FakeLLM()))
    db.commit()
    assert missing_folder_summaries(db) == []  # rows landed → no gap
    db.execute(text("DELETE FROM folder_summaries"))
    db.commit()
    assert missing_folder_summaries(db) == [("FSU", ""), ("FSU", "a")]  # emptied again


# ---------- on_progress hook (phase 98, task 01) ----------


def _record_progress() -> tuple[list[tuple[int, int, str, str]], Any]:
    """A fresh event list + the ``on_progress`` recorder that appends
    every ``(done, total, source, folder_path)`` event it receives."""
    events: list[tuple[int, int, str, str]] = []

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

    return events, record


def test_on_progress_fires_once_per_candidate_in_sorted_key_order(
    db: Session, clean_tables
) -> None:
    """Phase 98 (task 01): the hook fires once per candidate, in the
    same sorted ``(source, folder_path)`` order the folders are
    attempted — done climbs 1..total, total = the candidate count
    (the single-doc FSU-solo root IS a candidate at the ≥ 1 rule)."""
    _seed_catalogue(db)
    llm = _FakeLLM()
    events, record = _record_progress()
    stats = asyncio.run(generate_folder_summaries(db, llm, on_progress=record))
    assert events == [
        (1, 4, "FSU", ""),
        (2, 4, "FSU", "a"),
        (3, 4, "FSU", "a/b"),
        (4, 4, "FSU-solo", ""),
    ]
    assert llm.calls == 4  # one event per attempt, in the same order
    assert stats["generated"] == 4


def test_on_progress_manual_skip_still_advances(db: Session, clean_tables) -> None:
    """A manual-skip key is an INSTANT skip — no ``lite`` call burns,
    but the counter still advances for it (D5: the UI's position moves
    on either outcome)."""
    _seed_catalogue(db)
    db.add(
        FolderSummary(
            source="FSU", folder_path="a", summary="owner text",
            manually_edited=True,
        )
    )
    db.commit()
    llm = _FakeLLM()
    events, record = _record_progress()
    stats = asyncio.run(generate_folder_summaries(db, llm, on_progress=record))
    assert events == [
        (1, 4, "FSU", ""),
        (2, 4, "FSU", "a"),  # the instant manual skip still advances
        (3, 4, "FSU", "a/b"),
        (4, 4, "FSU-solo", ""),
    ]
    assert llm.calls == 3, "the skip itself burns no call"
    assert stats["kept_manual"] == 1 and stats["generated"] == 3


def test_on_progress_failed_key_still_advances(db: Session, clean_tables) -> None:
    """A failed (``LLMError``) key still advances — the hook fires
    BEFORE the attempt, so a fail-soft miss is visible to the UI as a
    completed step (the run never flips to failed, neither does the
    counter stall)."""
    _seed_catalogue(db)
    llm = _FakeLLM(fail_folders=("FSU/a",))
    events, record = _record_progress()
    stats = asyncio.run(generate_folder_summaries(db, llm, on_progress=record))
    assert events == [
        (1, 4, "FSU", ""),
        (2, 4, "FSU", "a"),  # the failed attempt still advances
        (3, 4, "FSU", "a/b"),
        (4, 4, "FSU-solo", ""),
    ]
    assert stats["failed"] == 1 and stats["generated"] == 3


def test_on_progress_only_missing_reports_the_missing_count(
    db: Session, clean_tables
) -> None:
    """Under ``only_missing=True`` ``total`` is the MISSING count (the
    loop-start count after the missing filter, not the full candidate
    count) and the events name exactly the missing keys, sorted."""
    _seed_catalogue(db)
    asyncio.run(generate_folder_summaries(db, _FakeLLM()))
    db.commit()
    db.execute(
        text(
            "DELETE FROM folder_summaries"
            " WHERE (source, folder_path) IN (('FSU', ''), ('FSU', 'a/b'))"
        )
    )
    db.commit()
    llm = _FakeLLM()
    events, record = _record_progress()
    stats = asyncio.run(
        generate_folder_summaries(db, llm, only_missing=True, on_progress=record)
    )
    assert events == [
        (1, 2, "FSU", ""),
        (2, 2, "FSU", "a/b"),
    ]
    assert llm.calls == 2
    assert stats["generated"] == 2


def test_on_progress_skip_true_fires_zero_calls(db: Session, clean_tables) -> None:
    """``skip=True`` (the ``--limit`` debug run) returns before the
    loop — the hook never fires."""
    _seed_catalogue(db)
    llm = _FakeLLM()
    events, record = _record_progress()
    stats = asyncio.run(generate_folder_summaries(db, llm, skip=True, on_progress=record))
    assert events == []
    assert llm.calls == 0
    assert stats == {"generated": 0, "failed": 0, "pruned": 0, "kept_manual": 0}


def test_on_progress_none_is_a_zero_cost_noop(db: Session, clean_tables) -> None:
    """``on_progress=None`` (the ``scripts/import_docs.py`` CLI path)
    behaves byte-identically to today: the fake LLM's call log, the
    stats dict, and the stored rows are identical across a hooked and
    a hook-less run."""
    _seed_catalogue(db)
    hooked = _FakeLLM()
    events, record = _record_progress()
    stats_hooked = asyncio.run(
        generate_folder_summaries(db, hooked, on_progress=record)
    )
    requests_hooked = hooked.requests.copy()
    assert len(events) == 4, "the hooked run fired (the contrast is real)"
    db.rollback()  # the generator only flushes — drop the uncommitted rows

    plain = _FakeLLM()
    stats_plain = asyncio.run(generate_folder_summaries(db, plain, on_progress=None))

    assert plain.requests == requests_hooked, "the fake LLM's call log is unchanged"
    assert stats_plain == stats_hooked
    assert _rows(db) == {
        ("FSU", ""): REPLY,
        ("FSU", "a"): REPLY,
        ("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"