brain-of-reese/tests/integration/test_regen_folder_summaries.py

"""Integration test: ``regen_folder_summaries`` runs the FULL-mode one-shot
regeneration (phase 155, task 02).

Drives ``scripts.regen_folder_summaries.main()`` end to end against the
local compose Postgres with a deterministic fake LLM (no live aipi,
fresh settings — the phase 31 integration's mocking style), over a
fixture catalogue (2 sources, 3 candidate folders: the ``Alpha`` source
root, ``Alpha/sub``, the ``Beta`` source root) whose ``folder_summaries``
rows are pre-stored with the formulaic pre-nudge voice and a backdated
``updated_at``:

- a full-mode run upserts EVERY candidate row — the text is the
  fake's deterministic reply for all of them and EVERY row's
  ``updated_at`` advances (a pure regen); the catalogue is
  unchanged, so the count-driven prune rule fires for nothing
  (``pruned=0``);
- the single cron-safe summary line logs the stats dict
  (``regen_folder_summaries: folders=… failed=… pruned=…
  kept_manual=… limit=…``);
- one ``lite`` call per candidate folder (a full regeneration burns a
  call per folder; the ``FOLDER_SUMMARY_MODE`` marker + the phase-155
  instruction are in the system prompt — the mock contract is intact
  end to end);
- a ``manually_edited`` row is SKIPPED (no ``lite`` burn on owner
  text, the row stays byte-identical — text AND ``updated_at``) and
  is named in the stats (``kept_manual=1``);
- a full ``lite`` failure is fail-soft PER FOLDER (phase 94): every
  previous row is kept (text AND ``updated_at``), the run's exit
  code stays 0, and the failure counts on the summary line
  (``folders=0 failed=3``) — the exit code is about the run (boot,
  DB, commit), never about per-folder model failures.
"""
from __future__ import annotations

import logging
from collections.abc import Iterator
from datetime import UTC, datetime, timedelta

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

from app.config import Settings
from app.db import db_available
from app.models import Document, FolderSummary
from app.rag.llm import LLMError
from scripts import regen_folder_summaries as regen
from tests.fakes import FakeEmbedder

#: The fixture catalogue: (source, path, title, summary) rows.
CATALOGUE: list[tuple[str, str, str, str]] = [
    ("Alpha", "a.md", "A One", "Summary of A"),
    ("Alpha", "b.md", "B Two", "Summary of B"),
    ("Alpha", "sub/c.md", "C Three", "Summary of C"),
    ("Beta", "x.md", "X One", "Summary of X"),
]

#: The candidate folders (recursive subtree ≥ 1 doc, per source):
#: the two source roots + the ``Alpha/sub`` folder.
CANDIDATES: list[tuple[str, str]] = [
    ("Alpha", ""),
    ("Alpha", "sub"),
    ("Beta", ""),
]

#: The formulaic pre-nudge voice the pre-stored rows carry (the
#: live catalogue's shape — what the one-shot exists to replace).
OLD_TEXT = "This folder contains the project's notes and setup guides."


class FailingChatEmbedder(FakeEmbedder):
    """A ``lite`` model that always fails (drives the per-folder
    fail-soft path)."""

    async def chat(self, messages: list[dict[str, str]], model: str | None = None) -> str:
        self.chat_calls.append(list(messages))
        raise LLMError("simulated lite-model failure (test sentinel)")


def _seed(db: Session, old: datetime) -> None:
    """Seed the fixture catalogue + the pre-stored folder-summary
    rows (the formulaic voice, backdated ``updated_at``)."""
    db.add_all(
        Document(
            source=source,
            path=path,
            full_path=f"/fixture/{source}/{path}",
            title=title,
            content=f"# {title}\n\nBody.\n",
            content_hash=f"{i:064d}"[:64],
            summary=summary,
        )
        for i, (source, path, title, summary) in enumerate(CATALOGUE)
    )
    db.add_all(
        FolderSummary(
            source=source,
            folder_path=folder_path,
            summary=OLD_TEXT,
            manually_edited=False,
            updated_at=old,
        )
        for source, folder_path in CANDIDATES
    )
    db.commit()


def _run_main(
    monkeypatch: pytest.MonkeyPatch,
    llm: FakeEmbedder,
    argv: list[str],
) -> tuple[int, list[str]]:
    """Run ``regen_folder_summaries.main`` with fresh settings and a
    fake LLM; returns the exit code + the module logger's summary
    lines (the single cron-safe line, PLAN §9)."""
    monkeypatch.setattr(
        regen, "get_settings", lambda: Settings(_env_file=None)  # pyright: ignore[reportCallIssue]
    )
    monkeypatch.setattr(regen, "LLMClient", lambda: llm)
    records: list[logging.LogRecord] = []

    class _Sink(logging.Handler):
        def emit(self, record: logging.LogRecord) -> None:
            records.append(record)

    sink = _Sink()
    log = logging.getLogger("scripts.regen_folder_summaries")
    log.addHandler(sink)
    log.setLevel(logging.INFO)
    try:
        rc = regen.main(argv)
    finally:
        log.removeHandler(sink)
    lines = [
        r.getMessage()
        for r in records
        if r.getMessage().startswith("regen_folder_summaries:")
    ]
    return rc, lines


@pytest.fixture(autouse=True)
def _clean_kb(db: Session) -> Iterator[None]:
    if not db_available():
        pytest.skip("Postgres not reachable — run `podman compose up -d db` first")
    db.execute(text("TRUNCATE chunks, documents, folder_summaries"))
    db.commit()
    yield
    db.execute(text("TRUNCATE chunks, documents, folder_summaries"))
    db.commit()


def test_full_regen_upserts_every_candidate_row(
    db: Session,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    old = datetime.now(UTC) - timedelta(days=1)
    _seed(db, old)
    llm = FakeEmbedder()

    rc, lines = _run_main(monkeypatch, llm, [])

    assert rc == 0
    # The single cron-safe summary line: the stats dict, one line.
    assert lines == [
        "regen_folder_summaries: folders=3 failed=0 pruned=0 "
        "kept_manual=0 limit=none"
    ]
    # A full regeneration burns exactly ONE lite call per candidate
    # folder (nothing else may touch chat).
    assert len(llm.chat_calls) == 3
    # The mock contract is intact end to end: the FOLDER_SUMMARY_MODE
    # marker + the phase-155 instruction (the two forbidden openers as
    # the constraint) are in every system prompt, and the user
    # message's first line names the folder (FOLDER_HEADER_PREFIX).
    expected_user_heads = ["Folder: Alpha", "Folder: Alpha/sub", "Folder: Beta"]
    for call, head in zip(llm.chat_calls, expected_user_heads, strict=True):
        by_role = {m["role"]: m["content"] for m in call}
        assert "FOLDER_SUMMARY_MODE" in by_role["system"]
        assert "Do NOT begin with 'This folder contains'" in by_role["system"]
        assert by_role["user"].splitlines()[0] == head
    # EVERY candidate row was upserted: the text is the fake's
    # deterministic reply (a pure function of the request — the same
    # "Summary of Folder:" for all three, the new voice for all of
    # them), and EVERY row's updated_at advanced past the backdate.
    db.expire_all()
    for source, folder_path in CANDIDATES:
        row = db.get(FolderSummary, (source, folder_path))
        assert row is not None
        assert row.summary == "Summary of Folder:"
        assert row.updated_at > old
    # No prune: the catalogue is unchanged, so the count-driven prune
    # rule fires for nothing — the row count is unchanged.
    assert db.execute(text("SELECT count(*) FROM folder_summaries")).scalar() == 3


def test_manual_row_is_kept_and_named_in_stats(
    db: Session,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """The owner's words are never regenerated (phase 97, task 01):
    the manual row stays byte-identical (text AND ``updated_at`` —
    no ``lite`` burn), the other candidate rows advance, and the skip
    is named in the stats (``kept_manual=1``)."""
    old = datetime.now(UTC) - timedelta(days=1)
    _seed(db, old)
    manual = db.get(FolderSummary, ("Alpha", ""))
    assert manual is not None
    manual.manually_edited = True
    manual.summary = "Owner words — the drill-down picture I wrote."
    db.commit()
    # Capture the manual row's stamp AFTER the seed (the column's
    # onupdate re-stamped it on the flag-setting UPDATE) — the
    # script must leave it byte-identical.
    manual = db.get(FolderSummary, ("Alpha", ""))
    assert manual is not None
    manual_stamp = manual.updated_at
    llm = FakeEmbedder()

    rc, lines = _run_main(monkeypatch, llm, [])

    assert rc == 0
    assert lines == [
        "regen_folder_summaries: folders=2 failed=0 pruned=0 "
        "kept_manual=1 limit=none"
    ]
    # No lite burn on owner text: only the two non-manual candidates.
    assert len(llm.chat_calls) == 2
    db.expire_all()
    row = db.get(FolderSummary, ("Alpha", ""))
    assert row is not None
    assert row.summary == "Owner words — the drill-down picture I wrote."
    assert row.updated_at == manual_stamp  # byte-identical — never re-stamped
    for source, folder_path in [("Alpha", "sub"), ("Beta", "")]:
        row = db.get(FolderSummary, (source, folder_path))
        assert row is not None
        assert row.summary == "Summary of Folder:"
        assert row.updated_at > old


def test_lite_failure_keeps_previous_rows_and_stays_exit_zero(
    db: Session,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """A full ``lite`` outage is fail-soft PER FOLDER (phase 94): the
    previous rows are kept (text AND ``updated_at`` — an old summary
    is better than none), the exit code stays 0 (the run itself was
    fine — boot, DB, commit all succeeded), and the failure counts on
    the summary line (``folders=0 failed=3``)."""
    old = datetime.now(UTC) - timedelta(days=1)
    _seed(db, old)
    llm = FailingChatEmbedder()

    rc, lines = _run_main(monkeypatch, llm, [])

    assert rc == 0  # a failed folder batch must not fail the run
    assert lines == [
        "regen_folder_summaries: folders=0 failed=3 pruned=0 "
        "kept_manual=0 limit=none"
    ]
    assert len(llm.chat_calls) == 3  # one attempt per candidate folder
    db.expire_all()
    for source, folder_path in CANDIDATES:
        row = db.get(FolderSummary, (source, folder_path))
        assert row is not None
        assert row.summary == OLD_TEXT  # previous row kept
        assert row.updated_at == old  # never re-stamped on failure