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

"""Unit: KB overview generator (phase 31, task 02).

The prompt tests are pure (no DB): ``KB_OVERVIEW_MODE`` system prompt,
per-document user lines (source/path/title/first summary line), and the
input cap with the shared ``[…truncated…]`` marker. The loader and
regenerator tests run against the local compose Postgres (preferred —
real single-row upsert), skipping with clear instructions when the stack
is not up — same pattern as ``test_importer.py``. The phase-162 section
pins the deterministic ``Sources: …`` inventory tail (LOCKED A2): the
format/order/counts, the 1-doc survival, the happy-path stored content,
the byte-identical zero-doc / ``LLMError`` paths, and the byte-identical
model input.
"""
from __future__ import annotations

import asyncio
import logging
import uuid
from datetime import UTC, datetime, timedelta

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

from app.config import Settings, get_settings
from app.models import Document, KbOverview
from app.rag.llm import LLMError
from app.rag.overview import (
    KB_OVERVIEW_MODE,
    OVERVIEW_INSTRUCTION,
    SYSTEM_PROMPT,
    _source_inventory,
    build_overview_prompt,
    load_kb_overview,
    regenerate_overview,
)
from app.rag.retriever import TRUNCATION_MARKER

REPLY = "- Homelab\n  - Kubernetes (k3s)\n- Deployments\n  - Borg backups"


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

    Records the messages and the ``model`` kwarg it was called with;
    returns a canned reply or raises (e.g. :class:`LLMError`).
    """

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

    async def chat(
        self, messages: list[dict[str, str]], model: str | None = None
    ) -> str:
        self.calls += 1
        self.messages = list(messages)
        self.model = model
        if self._fail is not None:
            raise self._fail
        return self._reply


# ---------- build_overview_prompt: system ----------


def test_system_prompt_has_marker_and_locked_instruction() -> None:
    assert SYSTEM_PROMPT.startswith(KB_OVERVIEW_MODE)
    assert OVERVIEW_INSTRUCTION in SYSTEM_PROMPT
    for fragment in (
        "compact plain-text outline",
        "basic categories and topics",
        "Group by source",
        "use `-` bullet lines",
        "~1500 characters",
        "no markdown headings",
        "no topics not present in the list",
    ):
        assert fragment in SYSTEM_PROMPT
    system, _ = build_overview_prompt([])
    assert system == SYSTEM_PROMPT
    assert KB_OVERVIEW_MODE in system  # the marker the E2E mock keys on


# ---------- build_overview_prompt: user lines ----------


def test_user_lines_carry_source_path_title_and_first_summary_line() -> None:
    rows = [
        ("Homelab", "kubernetes/k3s.md", "K3s Cluster", None),
        (
            "Deployments",
            "backups/borg.yaml",
            "Borg Backups",
            "Backups run nightly via the borg schedule.\nSource: Deployments/backups/borg.yaml",
        ),
    ]
    system, user = build_overview_prompt(rows)
    assert system == SYSTEM_PROMPT
    assert user == (
        "Homelab — kubernetes/k3s.md — K3s Cluster\n"
        "Deployments — backups/borg.yaml — Borg Backups — "
        "Backups run nightly via the borg schedule."
    )


def test_user_line_omits_summary_field_when_absent() -> None:
    _, user = build_overview_prompt([("Homelab", "a.md", "Title A", None)])
    assert user == "Homelab — a.md — Title A"
    assert not user.endswith(" — ")  # no dangling dash


def test_user_line_omits_summary_field_when_blank() -> None:
    _, user = build_overview_prompt([("Homelab", "a.md", "Title A", "  \n\t ")])
    assert user == "Homelab — a.md — Title A"


def test_user_line_uses_only_first_summary_line() -> None:
    """Multi-line summaries contribute their first line only (the
    model-written lead sentence; the ``Source: …`` pointer is last)."""
    _, user = build_overview_prompt(
        [
            (
                "Homelab",
                "a.yaml",
                "Title A",
                "First line.\nSecond line.\nSource: Homelab/a.yaml",
            )
        ]
    )
    assert user == "Homelab — a.yaml — Title A — First line."
    assert "Second line" not in user
    assert "Source:" not in user


def test_zero_rows_yield_empty_user() -> None:
    _, user = build_overview_prompt([])
    assert user == ""


# ---------- build_overview_prompt: input cap ----------


def test_user_prompt_truncated_with_marker_when_over_custom_cap() -> None:
    rows = [("S", f"p{i}.md", f"T{i}", None) for i in range(10)]
    _, full = build_overview_prompt(rows, max_chars=10_000)
    cap = 25
    _, user = build_overview_prompt(rows, 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:
    rows = [("S", "p.md", "T", None)]  # "S — p.md — T" = 12 chars
    _, user = build_overview_prompt(rows, max_chars=12)
    assert user == "S — p.md — T"
    assert TRUNCATION_MARKER not in user


def test_user_prompt_truncated_at_default_cap() -> None:
    """No explicit cap → ``BOR_OVERVIEW_INPUT_MAX_CHARS`` (read from the
    live settings, so the test holds for any configured value)."""
    cap = get_settings().overview_input_max_chars
    rows = [("S", f"p{i}.md", "T", None) for i in range(3_000)]
    _, user = build_overview_prompt(rows)
    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 "p2999.md" not in body  # the overflow never reaches the model


# ---------- phase 142, task 01: cap-ordering repro -----------------------


def _phase_dir_rows() -> list[tuple[str, str, str, str]]:
    """20 rows in the owner's KB phase-directory shape (task 01).

    ``01_old_a.md`` … ``10_old_j.md`` are the OLDER documents and
    ``11_new_k.md`` … ``20_new_t.md`` the NEWER ones (the suffix is the
    i-th letter, a–t), but the zero-padded
    numbering makes the newest sort lexicographically LAST — the owner's
    ``113_…`` vs ``128_…`` vs ``131_…`` shape (``131_…`` sorts after
    ``13_…``, so the newest phases trail the lexicographic list). Each
    rendered line (``source — path — title — summary``) is exactly 100
    chars, so a 500-char cap fits five lines plus four newlines (504 —
    one line over budget): the cut is guaranteed to land mid-list, and
    the lexicographic first (the oldest) vs last (the newest) contrast
    is what the repro asserts on.
    """
    rows: list[tuple[str, str, str, str]] = []
    for i in range(1, 21):
        kind = "old" if i <= 10 else "new"
        path = f"{i:02d}_{kind}_{chr(ord('a') + i - 1)}.md"
        title = f"Phase {i:02d}"
        prefix = f"s — {path} — {title} — "
        rows.append(("s", path, title, "x" * (100 - len(prefix))))
    return rows


def test_cap_keeps_the_given_prefix_and_cuts_the_tail() -> None:
    """Phase 142, task 01 — the cap behavior pinned AS-IS (green before
    AND after the fix — the cap itself is unchanged, only the ordering
    of the list feeding it is). Given rows in lexicographic order
    (``01_old_a.md`` first, ``20_new_t.md`` last), a 500-char cap keeps
    exactly the lexicographic FIRST lines and drops the tail: ``01_old_a``
    survives, ``20_new_t`` is cut, and the shared marker makes the cut
    visible on its own line (the current behavior, byte-for-byte).
    """
    rows = _phase_dir_rows()
    _, full = build_overview_prompt(rows, max_chars=10_000)
    cap = 500
    _, user = build_overview_prompt(rows, max_chars=cap)
    assert "01_old_a.md" in user  # the lexicographic prefix survives the cut
    assert "20_new_t.md" not in user  # the lexicographic tail never reaches the model
    assert user.endswith("\n" + TRUNCATION_MARKER)  # the cut is visible
    body = user.removesuffix("\n" + TRUNCATION_MARKER)
    assert len(body) == cap  # cut exactly at the cap
    assert body == full[:cap]  # a plain prefix cut of the joined list


# ---------- load_kb_overview ----------


def test_load_kb_overview_no_row_empty(db: Session) -> None:
    db.execute(text("DELETE FROM kb_overview"))
    db.commit()
    assert load_kb_overview(db) == ""


def test_load_kb_overview_returns_trimmed_content(db: Session) -> None:
    db.execute(text("DELETE FROM kb_overview"))
    db.add(KbOverview(id=1, content=f"  {REPLY}  \n"))
    db.commit()
    assert load_kb_overview(db) == REPLY


def test_load_kb_overview_whitespace_only_row_empty(db: Session) -> None:
    db.execute(text("DELETE FROM kb_overview"))
    db.add(KbOverview(id=1, content="  \n\t "))
    db.commit()
    assert load_kb_overview(db) == ""


# ---------- regenerate_overview ----------


def _add_doc(
    db: Session,
    source: str,
    path: str,
    title: str,
    summary: str | None = None,
    created_at: datetime | 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,
    )
    if created_at is not None:
        doc.created_at = created_at
    db.add(doc)
    db.commit()
    return doc


def _truncate_documents(db: Session) -> None:
    db.execute(text("TRUNCATE chunks, documents"))
    db.commit()


def _overview_row(db: Session) -> KbOverview | None:
    return db.get(KbOverview, 1)


@pytest.fixture()
def clean_overview(db: Session):
    db.execute(text("DELETE FROM kb_overview"))
    db.commit()
    yield
    db.execute(text("DELETE FROM kb_overview"))
    db.commit()


def test_regenerate_happy_path_creates_single_row_id_1(
    db: Session, clean_overview, caplog: pytest.LogCaptureFixture
) -> None:
    _truncate_documents(db)
    try:
        _add_doc(db, "Homelab", "b.md", "B")
        _add_doc(db, "Homelab", "a.md", "A", "A summary.\nSource: Homelab/a.md")
        llm = _FakeLLM()
        with caplog.at_level(logging.INFO, logger="app.rag.overview"):
            ok = asyncio.run(regenerate_overview(llm, db))
            assert ok is True
            assert llm.calls == 1
            row = db.get(KbOverview, 1)
            assert row is not None, "the single row must be upserted"
            assert row.id == 1
            # Phase 162 (LOCKED A2): the stored content carries the
            # deterministic source-inventory tail after the model text.
            assert row.content == REPLY + "\nSources: Homelab (2 docs)"
            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"
            assert f"overview: regenerated docs=2 chars={len(row.content)}" in caplog.text
    finally:
        _truncate_documents(db)


def test_regenerate_updates_existing_row_in_place(db: Session, clean_overview) -> None:
    _truncate_documents(db)
    past = datetime(2020, 1, 1, 12, 0, 0, tzinfo=UTC)
    db.add(KbOverview(id=1, content="old outline", updated_at=past))
    db.commit()
    try:
        _add_doc(db, "Homelab", "a.md", "A")
        ok = asyncio.run(regenerate_overview(_FakeLLM(), db))
        assert ok is True
        count = db.scalar(text("SELECT count(*) FROM kb_overview"))
        assert count == 1, "still exactly one row after the re-generation"
        row = db.get(KbOverview, 1)
        assert row is not None
        assert row.content == REPLY + "\nSources: Homelab (1 doc)"  # replaced, not appended
        assert row.updated_at is not None and row.updated_at > past
    finally:
        _truncate_documents(db)


def test_regenerate_orders_documents_newest_first_then_source_path(
    db: Session, clean_overview
) -> None:
    """Phase 142, task 02 — the generator's document list is RECENT-FIRST:
    ``created_at`` descending, with the ``(source, path)`` tie-break
    keeping same-date documents in the pre-phase-142 lexicographic
    order. Six documents in two ``created_at`` batches (inserted out of
    order, sources interleaved across the batches): the SELECT must
    return the NEWER batch first, ``(source, path)`` within each batch —
    and the date order must dominate the source/path order (pre-fix the
    plain ``order_by(source, path)`` put ``S1 — a.md — Old S1a`` first)."""
    _truncate_documents(db)
    try:
        old = datetime(2026, 1, 1, tzinfo=UTC)
        new = datetime(2026, 6, 1, tzinfo=UTC)
        # The NEWER batch (2026-06-01) — inserted in a jumbled order.
        _add_doc(db, "S2", "z.md", "New S2z", created_at=new)
        _add_doc(db, "S1", "y.md", "New S1y", created_at=new)
        _add_doc(db, "S2", "a.md", "New S2a", created_at=new)
        # The OLDER batch (2026-01-01) — also jumbled.
        _add_doc(db, "S2", "b.md", "Old S2b", created_at=old)
        _add_doc(db, "S1", "c.md", "Old S1c", created_at=old)
        _add_doc(db, "S1", "a.md", "Old S1a", created_at=old)
        llm = _FakeLLM()
        assert asyncio.run(regenerate_overview(llm, db)) is True
    finally:
        _truncate_documents(db)

    user = next(m["content"] for m in llm.messages if m["role"] == "user")
    # Newest batch first (all three 2026-06-01 docs before any 2026-01-01
    # doc), (source, path) within a batch — the pre-fix lexicographic
    # order (S1/a, S1/c, S1/y, S2/a, S2/b, S2/z) would fail this pin.
    assert user == (
        "S1 — y.md — New S1y\n"
        "S2 — a.md — New S2a\n"
        "S2 — z.md — New S2z\n"
        "S1 — a.md — Old S1a\n"
        "S1 — c.md — Old S1c\n"
        "S2 — b.md — Old S2b"
    )


def test_document_list_reaching_the_model_leads_with_newest_docs(
    db: Session, clean_overview
) -> None:
    """Phase 142, task 01 repro, flipped green by task 02 (the strict
    xfail marker is removed — a pass is now a pass).

    The document list reaching the lite model must LEAD with the NEWEST
    rows (``created_at`` desc — the document's date, phase 106) so that,
    when the cap bites, the newest documents — not the lexicographically
    first — are the ones the outline covers. Pre-fix the query was plain
    ``order_by(source, path)``: with the phase-dir rows below (recency
    strictly OPPOSITE to the path order) that put ``01_old_a.md`` — the
    OLDEST — first and ``20_new_t.md`` — the NEWEST — last, the traced
    "summaries only up to phase 113" failure at unit scale. The
    assertion is the POST-FIX order task 02's ``created_at``-desc
    ordering delivers.
    """
    _truncate_documents(db)
    try:
        base = datetime(2026, 1, 1, tzinfo=UTC)
        for i, (source, path, title, summary) in enumerate(_phase_dir_rows()):
            _add_doc(db, source, path, title, summary)
            # Row i (0-based) is i days after the base: index 0
            # (``01_old_a.md``) is the OLDEST, index 19 (``20_new_t.md``)
            # the NEWEST — recency strictly opposite to the path order.
            db.execute(
                text("UPDATE documents SET created_at = :d WHERE path = :p"),
                {"d": base + timedelta(days=i), "p": path},
            )
            db.commit()
        llm = _FakeLLM()
        assert asyncio.run(regenerate_overview(llm, db)) is True
    finally:
        _truncate_documents(db)

    user = next(m["content"] for m in llm.messages if m["role"] == "user")
    lines = user.splitlines()
    assert len(lines) == 20  # no cap at this size — the ORDER is what is asserted
    assert lines[0].startswith("s — 20_new_t.md"), (
        "the list must lead with the NEWEST doc (20_new_t) — pre-fix it "
        "leads with the lexicographic prefix (01_old_a, the oldest)"
    )
    assert lines[-1].startswith("s — 01_old_a.md"), (
        "the oldest doc must sort LAST — pre-fix it sorts FIRST"
    )


def test_regenerate_calls_the_configured_summary_model(
    db: Session, clean_overview
) -> None:
    _truncate_documents(db)
    try:
        _add_doc(db, "Homelab", "a.md", "A")
        llm = _FakeLLM()
        assert asyncio.run(regenerate_overview(llm, db)) is True
    finally:
        _truncate_documents(db)

    assert llm.model == llm.settings.llm_summary_model  # the ``lite`` default
    assert llm.model == "lite"
    assert [m["role"] for m in llm.messages] == ["system", "user"]
    assert KB_OVERVIEW_MODE in llm.messages[0]["content"]
    assert "A" in llm.messages[1]["content"]


def test_regenerate_zero_docs_leaves_existing_row_untouched(
    db: Session, clean_overview
) -> None:
    db.execute(text("DELETE FROM kb_overview"))
    db.add(KbOverview(id=1, content="old outline"))
    db.commit()
    _truncate_documents(db)
    llm = _FakeLLM()
    ok = asyncio.run(regenerate_overview(llm, db))
    assert ok is False
    assert llm.calls == 0, "no KB → no wasted lite-model call"
    row = db.get(KbOverview, 1)
    assert row is not None and row.content == "old outline"  # untouched


def test_regenerate_zero_docs_without_row_creates_nothing(db: Session, clean_overview) -> None:
    _truncate_documents(db)
    llm = _FakeLLM()
    ok = asyncio.run(regenerate_overview(llm, db))
    assert ok is False
    assert llm.calls == 0
    assert _overview_row(db) is None, "no row must be invented for an empty KB"


def test_regenerate_llm_error_leaves_previous_row_intact(
    db: Session, clean_overview, caplog: pytest.LogCaptureFixture
) -> None:
    db.execute(text("DELETE FROM kb_overview"))
    db.add(KbOverview(id=1, content="old outline"))
    db.commit()
    _truncate_documents(db)
    try:
        _add_doc(db, "Homelab", "a.md", "A")
        llm = _FakeLLM(fail=LLMError("simulated lite-model failure"))
        with caplog.at_level(logging.ERROR, logger="app.rag.overview"):
            ok = asyncio.run(regenerate_overview(llm, db))
            assert ok is False, "fail-soft: the import must not be dragged down"
            row = db.get(KbOverview, 1)
            assert row is not None and row.content == "old outline", (
                "the previous outline stays — an old outline is better than none"
            )
            assert "overview: regeneration failed —" in caplog.text
            assert "simulated lite-model failure" in caplog.text
    finally:
        _truncate_documents(db)


def test_regenerate_llm_error_without_row_creates_nothing(
    db: Session, clean_overview
) -> None:
    _truncate_documents(db)
    try:
        _add_doc(db, "Homelab", "a.md", "A")
        llm = _FakeLLM(fail=LLMError("simulated lite-model failure"))
        ok = asyncio.run(regenerate_overview(llm, db))
    finally:
        _truncate_documents(db)

    assert ok is False
    assert _overview_row(db) is None


def test_regenerate_opens_own_session_when_none(db: Session, clean_overview) -> None:
    """``session=None`` → a private ``SessionLocal`` session is opened,
    committed, and closed (the import-script call path)."""
    _truncate_documents(db)
    try:
        _add_doc(db, "Homelab", "a.md", "A")
        ok = asyncio.run(regenerate_overview(_FakeLLM()))
    finally:
        _truncate_documents(db)

    assert ok is True
    row = db.get(KbOverview, 1)  # fresh lookup via the test session
    assert row is not None and row.content == REPLY + "\nSources: Homelab (1 doc)"


# ---------- phase 162: the deterministic source inventory (LOCKED A2) -------


def test_source_inventory_format_counts_and_catalog_order() -> None:
    """Exact format, singular/plural, and catalog order: one entry per
    DISTINCT source — ``<name> (<k> doc|docs)`` joined with ``, ``
    behind ``Sources: ``, ordered ``source, path`` (the scrambled
    input rows prove the catalog order is the OUTPUT order, not the
    recency-first order of the rows the generator reads)."""
    rows = [
        ("b", "x.md", "X", None),
        ("a", "y.md", "Y", None),
        ("c", "p1.md", "P1", None),
        ("b", "z.md", "Z", None),
        ("c", "p2.md", "P2", None),
        ("c", "p3.md", "P3", None),
    ]
    assert _source_inventory(rows) == "Sources: a (1 doc), b (2 docs), c (3 docs)"


def test_source_inventory_keeps_a_one_doc_source_visible() -> None:
    """The 2026-10-07 ``ikea_instructions`` case: a 1-doc source among
    many appears with ``(1 doc)`` — no source is invisible to
    ``<knowledge_base>``, whatever the ``lite`` outline says."""
    rows = [
        ("gufo", "a.md", "A", None),
        (
            "ikea_instructions",
            "besta-frame-white__AA-1272068-11-100.md",
            "Assembly Instructions",
            None,
        ),
        ("gufo", "b.md", "B", None),
        ("players_handbook", "core.md", "Core", None),
    ]
    assert _source_inventory(rows) == (
        "Sources: gufo (2 docs), ikea_instructions (1 doc), "
        "players_handbook (1 doc)"
    )


def test_regenerate_appends_the_inventory_tail_to_the_stored_content(
    db: Session, clean_overview, caplog: pytest.LogCaptureFixture
) -> None:
    """Happy path, exact pin on the fake's fixed output: the stored row
    = model text (trailing newlines rstripped) + ``\\n`` + the inventory
    line; the log's chars = the FULL stored content, tail included."""
    _truncate_documents(db)
    try:
        _add_doc(db, "players_handbook", "rpg.md", "RPG")
        _add_doc(db, "Homelab", "b.md", "B")
        _add_doc(db, "Homelab", "a.md", "A")
        llm = _FakeLLM(reply=REPLY + "\n\n")
        expected = REPLY + "\nSources: Homelab (2 docs), players_handbook (1 doc)"
        with caplog.at_level(logging.INFO, logger="app.rag.overview"):
            ok = asyncio.run(regenerate_overview(llm, db))
            assert ok is True
            row = db.get(KbOverview, 1)
            assert row is not None
            assert row.content == expected  # no blank line before the tail
            assert f"overview: regenerated docs=3 chars={len(expected)}" in caplog.text
    finally:
        _truncate_documents(db)


def test_regenerate_model_input_never_gains_the_inventory_tail(
    db: Session, clean_overview
) -> None:
    """The model input is byte-identical to the pre-phase prompt: the
    user message the fake received equals ``build_overview_prompt`` for
    the same catalogue rows — the tail is never sent to ``lite``."""
    _truncate_documents(db)
    try:
        _add_doc(db, "Homelab", "a.md", "A", "A summary.\nSource: Homelab/a.md")
        _add_doc(db, "Deployments", "b.yaml", "B")
        llm = _FakeLLM()
        assert asyncio.run(regenerate_overview(llm, db)) is True
        rows = db.execute(
            select(Document.source, Document.path, Document.title, Document.summary)
            .order_by(Document.created_at.desc(), Document.source, Document.path)
        ).all()
    finally:
        _truncate_documents(db)

    user = next(m["content"] for m in llm.messages if m["role"] == "user")
    _, expected_user = build_overview_prompt([tuple(r) for r in rows])
    assert user == expected_user
    assert "Sources:" not in user


def test_regenerate_zero_docs_keeps_tailless_pre_phase_row(
    db: Session, clean_overview
) -> None:
    """Zero-doc early return: byte-identical pre-phase behavior — the
    pre-phase (tail-less) row survives untouched, no tail is invented."""
    db.execute(text("DELETE FROM kb_overview"))
    pre_phase_row = "- Old outline\n- from before phase 162"
    db.add(KbOverview(id=1, content=pre_phase_row))
    db.commit()
    _truncate_documents(db)
    llm = _FakeLLM()
    ok = asyncio.run(regenerate_overview(llm, db))
    assert ok is False
    assert llm.calls == 0
    row = db.get(KbOverview, 1)
    assert row is not None and row.content == pre_phase_row
    assert "Sources:" not in row.content


def test_regenerate_llm_error_preserves_an_existing_inventory_tail(
    db: Session, clean_overview
) -> None:
    """The tail is CONTENT: a previous row that already carried a
    ``Sources: …`` tail survives an ``LLMError`` byte-identically —
    including the tail (an old outline is better than none)."""
    db.execute(text("DELETE FROM kb_overview"))
    previous = "- Old outline\nSources: Homelab (3 docs)"
    db.add(KbOverview(id=1, content=previous))
    db.commit()
    _truncate_documents(db)
    try:
        _add_doc(db, "Homelab", "a.md", "A")
        llm = _FakeLLM(fail=LLMError("simulated lite-model failure"))
        ok = asyncio.run(regenerate_overview(llm, db))
        assert ok is False
        row = db.get(KbOverview, 1)
        assert row is not None and row.content == previous
    finally:
        _truncate_documents(db)