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

"""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,
    build_folder_summary_prompt,
    folder_of,
    folder_summary_table_empty,
    generate_folder_summaries,
    group_by_folder,
    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 ``Folder: …`` header
    (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))
        for folder in self._fail_folders:
            if FOLDER_HEADER_PREFIX + folder in user:
                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 (≥ 1 docs): the ≥ 2 rule is
    the GENERATOR's (the recursive count below the minimum yields no
    row — 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]


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


# ---------- 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)."""
    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)) == (
        "Fixture folder summary 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)) == (
        "Fixture folder summary for Homelab."
    )


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 == "Fixture folder summary 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 — all ≥ the minimum); FSU-solo has one
    doc (its root folder is below the minimum — no row, no call)."""
    _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 ≥ 2 recursive docs gets a row (the source root
    row included — ``folder_path = ''``); single-doc folders get none;
    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": 3, "failed": 0, "pruned": 0}
    assert llm.calls == 3, "one lite call per candidate folder (the solo folder: none)"

    stored = _rows(db)
    assert set(stored) == {("FSU", ""), ("FSU", "a"), ("FSU", "a/b")}
    assert all(summary == REPLY for summary in stored.values())
    assert ("FSU-solo", "") not in stored, (
        "a single-doc folder is fully described by its one file line — no row"
    )

    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",
    ]
    # 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=3 failed=0 pruned=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": 2, "failed": 1, "pruned": 0}
    assert llm.calls == 3  # 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, (
        "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 dropped below 2 recursive docs are deleted
    (pruned/renamed — the summary would go stale); rows for folders
    that still qualify persist (an unchanged folder's summary is still
    true — regenerated in place)."""
    _seed_catalogue(db)
    # A stale row for a folder no longer in the catalogue (3→1 docs /
    # renamed away) + a live row 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"] == 2  # gone/old + the FSU-solo root (1 doc)

    stored = _rows(db)
    assert ("FSU", "gone/old") not in stored, "the stale folder row must be pruned"
    assert ("FSU-solo", "") not in stored, (
        "a folder that dropped below 2 docs loses its row"
    )
    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}
    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}
    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 == 2  # the ≥ 2 scope rule, pinned by name


def test_folder_summary_table_empty_gate(db: Session, clean_tables) -> None:
    """The sync-path gate probe (phase 94, task 02): empty → True
    (the first full sync after migration 0017 must still generate),
    one row → False (a populated table waits for a KB change)."""
    assert folder_summary_table_empty(db) is True  # the truncated table
    _add_doc(db, "FSU", "a/one.md", "One")
    _add_doc(db, "FSU", "a/two.md", "Two")
    asyncio.run(generate_folder_summaries(db, _FakeLLM()))
    db.commit()
    assert folder_summary_table_empty(db) is False  # rows landed
    db.execute(text("DELETE FROM folder_summaries"))
    db.commit()
    assert folder_summary_table_empty(db) is True  # emptied again