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

"""Integration: phase 194 — the pre-scan predicts the change set, off the loop.

The owner report that opened the phase (2026-10-19, verbatim): *"I just synced
the prod brain of reese and it said '1216 remaining' when it only had like 10
files to sync."* Prod holds ~10,000 importable files and that run changed ~10.
The phase-193 ``N remaining`` label was arithmetically fine — it was fed the
WALK (every importable file) instead of the WORK. The fix lives in
``app/rag/importer.py``: before the walk, ``_prescan_pending`` hashes every
importable file once and reads ``documents.file_hash`` once per file — the
exact predicate the phase-147 unchanged fast path applies — and the per-file
``progress`` hook reports the position in THAT set.

What this file measures with a real ``import_sources`` run:

* **the parity that makes the number honest** — on a stable knowledge base
  the predicted set's size equals the run's terminal ``added + updated``: the
  fresh arm (everything pending, exactly what a first import indexes), the
  unchanged arm (nothing pending, so no count part at all), and the
  incremental arm (one changed file of four — the owner's prod case, where
  the walk is 4 and the work is 1);
* **zero entity hydration** — the scan's statement projects exactly
  ``documents.file_hash`` (no other column, no ORM row), so a 10,000-file
  scan retains nothing: the run's session is still empty at the first
  ``progress`` call, before the processing loop has loaded any row (phase
  184/185);
* **the span is OFF the loop** — with a stubbed hash that blocks, a heartbeat
  coroutine keeps ticking through the scan and the hash runs on a different
  thread than the one that awaited it (the phase-179 pattern of
  ``test_sync_git_freeze.py``: the git clone used to park the single uvicorn
  loop and freeze the ``GET /api/sync/status`` poll the UI runs every 2 s,
  and a 10,000-file hash+SELECT span inline is the same defect class);
* **the cancel contract** — cancelling the run at the scan's await lands the
  cancellation immediately, while the worker itself runs to completion
  (threads are not interruptible, phase 134) and nothing gets indexed.

The CLI and the upload job pass no ``progress``, so they run no scan at all;
their byte-identity is pinned in ``tests/unit/test_sync_pending_count.py``.
"""
from __future__ import annotations

import asyncio
import re
import threading
import time
from collections.abc import Callable, Iterator
from pathlib import Path
from typing import Any, NamedTuple

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

import app.rag.importer as importer
from app.db import engine
from app.models import Document
from app.rag.importer import ImportSummary, import_sources
from tests.fakes import FakeEmbedder

#: The probe's own source-name prefix — nothing else in the KB uses it.
PREFIX = "phase194-it-"

#: A blocking hash of 60 ms per file: four files is a ~250 ms scan — long
#: enough that a heartbeat polling every 5 ms cannot tick during it if the
#: span ever runs on the loop thread, short enough that the suite stays fast.
HASH_BLOCK_S = 0.06

#: The heartbeat tick count that proves the loop was free across ONE scanned
#: file (60 ms / 5 ms ≈ 12 ticks; a blocked loop ticks 0). The ceiling-side
#: assertion is the same number generous: this is loop-freeness, not timing.
MIN_TICKS_PER_FILE = 4


class ProgressCall(NamedTuple):
    """One ``progress(source, rel, done, total)`` report."""

    source: str
    rel: str
    done: int
    total: int


class Run(NamedTuple):
    """One real ``import_sources`` pass and everything it reported."""

    summary: ImportSummary
    progress: list[ProgressCall]
    scan: list[tuple[int, int]]


def _llm(**setting_overrides: Any) -> FakeEmbedder:
    """The deterministic in-process fake on the SHIPPED defaults (the
    pending-count switch is therefore ON — the phase's default)."""
    from app.config import Settings

    llm = FakeEmbedder()
    llm.settings = Settings(_env_file=None, **setting_overrides)  # pyright: ignore[reportCallIssue]
    return llm


def _root(tmp_path: Path, name: str, files: dict[str, str]) -> Path:
    """One source root (the ``file://``-style temp dir of the house
    import fixtures) with *files* under it."""
    assert name.startswith(PREFIX), f"the probe source must be {PREFIX}-prefixed"
    root = tmp_path / name
    for rel, body in files.items():
        target = root / rel
        target.parent.mkdir(parents=True, exist_ok=True)
        target.write_text(body, encoding="utf-8")
    return root


def _run(
    session: Session,
    *roots: Path,
    llm: FakeEmbedder | None = None,
    **kwargs: Any,
) -> Run:
    """One real import over *roots*, collecting both position hooks."""
    progress: list[ProgressCall] = []
    scan: list[tuple[int, int]] = []
    summary = asyncio.run(
        import_sources(
            list(roots),
            llm or _llm(),
            session=session,
            progress=lambda s, r, d, t: progress.append(ProgressCall(s, r, d, t)),
            on_scan_progress=lambda p, t: scan.append((p, t)),
            **kwargs,
        )
    )
    return Run(summary, list(progress), list(scan))


@pytest.fixture()
def statement_capture() -> Iterator[list[str]]:
    """Every statement the engine executes for the duration of ONE test.

    The ``before_cursor_execute`` technique of phase 183/185
    (``tests/integration/test_import_fast_path_columns.py``): what a SELECT
    actually projects is invisible in the ORM's Python API and plain in the
    statement it compiles, so a capture is the only way to pin "the pre-scan
    reads one column". Removed in teardown, so no other suite sees it.
    """
    captured: list[str] = []

    def _listener(
        conn: object,  # noqa: ARG001
        cursor: object,  # noqa: ARG001
        statement: str,
        parameters: object,  # noqa: ARG001
        context: object,  # noqa: ARG001
        executemany: bool,  # noqa: ARG001
    ) -> None:
        captured.append(statement)

    event.listen(engine, "before_cursor_execute", _listener)
    try:
        yield captured
    finally:
        event.remove(engine, "before_cursor_execute", _listener)


@pytest.fixture(autouse=True)
def _clean_probe_documents(db: Session) -> Iterator[None]:
    """``documents``/``chunks`` are global state — the probe's rows are
    removed before and after every test (the ``test_sync_api.py`` hygiene
    pattern, narrowed to this module's own source names)."""
    _drop(db)
    yield
    _drop(db)


def _drop(db: Session) -> None:
    db.execute(text("DELETE FROM chunks WHERE document_id IN (SELECT id FROM "
                    "documents WHERE source LIKE :p)"), {"p": f"{PREFIX}%"})
    db.execute(text("DELETE FROM documents WHERE source LIKE :p"), {"p": f"{PREFIX}%"})
    db.commit()
    db.expunge_all()


# --------------------------------------------------------------------------
# The parity: the predicted set IS the work
# --------------------------------------------------------------------------

_FOUR_FILES = {
    "a.md": "# A\n\nalpha content\n",
    "b.md": "# B\n\nbravo content\n",
    "c.md": "# C\n\ncharlie content\n",
    "notes/d.md": "# D\n\ndelta content\n",
}


def test_fresh_kb_pending_set_equals_the_added_count(
    db: Session, tmp_path: Path
) -> None:
    """Arm 1 — the first import: walk == work, so the numbers are the
    phase-193 ones (A1's degenerate parity, the reason the phase-193 E2E
    suite never saw the defect).

    Every file has no row, so every file is pending: the hook reports
    ``1/4 … 4/4`` and the terminal says ``added=4`` — the predicted size
    equals ``added + updated`` exactly."""
    root = _root(tmp_path, f"{PREFIX}fresh", _FOUR_FILES)

    run = _run(db, root)

    assert (run.summary.files, run.summary.added, run.summary.errors) == (4, 4, 0)
    assert [(c.done, c.total) for c in run.progress] == [(1, 4), (2, 4), (3, 4), (4, 4)]
    assert run.scan == [(1, 4), (2, 4), (3, 4), (4, 4)]
    assert run.progress[-1].total == run.summary.added + run.summary.updated


def test_unchanged_kb_predicts_nothing_and_terminal_says_no_work(
    db: Session, tmp_path: Path
) -> None:
    """Arm 2 — the no-change sync: nothing pending, so no count part.

    The scan walks all four files and predicts none of them; every ``progress``
    call therefore carries ``(0, 0)``, which the phase-64 ``total === 0``
    guard renders as a bare label with no ``N remaining`` — the honest shape
    for a run with no work, where the pre-phase label said ``4 remaining``
    while walking. ``summary.files`` (the log's walked-file count) is
    untouched: the run still processed 4 files and still logged 4 unchanged.
    """
    root = _root(tmp_path, f"{PREFIX}unchanged", _FOUR_FILES)
    seed = _run(db, root)
    assert seed.summary.added == 4

    run = _run(db, root)

    assert (run.summary.files, run.summary.added, run.summary.updated) == (4, 0, 0)
    assert run.summary.unchanged == 4
    assert [(c.done, c.total) for c in run.progress] == [(0, 0)] * 4
    # The scan still walked (and reported) every file — the DENOMINATOR of the
    # scan is the importable count, not the prediction.
    assert run.scan == [(1, 4), (2, 4), (3, 4), (4, 4)]
    assert run.summary.added + run.summary.updated == 0


def test_incremental_kb_counts_the_changed_files_only(
    db: Session, tmp_path: Path
) -> None:
    """Arm 3 — the owner's prod case at 1-of-4 scale.

    One changed file out of four: the walk is still 4 (``summary.files``, the
    log line, and every ``progress`` call naming a file), but the count is the
    WORK — ``total`` 1, the position climbing only when the walk reaches the
    changed file, and holding at 0 while it moves over the unchanged ones.
    The parity pin: the prediction's size equals the terminal's
    ``added + updated``.
    """
    root = _root(tmp_path, f"{PREFIX}incremental", _FOUR_FILES)
    seed = _run(db, root)
    assert seed.summary.added == 4

    (root / "notes" / "d.md").write_text(
        "# D\n\ndelta content EDITED on disk\n", encoding="utf-8"
    )

    run = _run(db, root)

    assert (run.summary.files, run.summary.updated, run.summary.unchanged) == (
        4,
        1,
        3,
    )
    assert [(c.rel, c.done, c.total) for c in run.progress] == [
        ("a.md", 0, 1),
        ("b.md", 0, 1),
        ("c.md", 0, 1),
        ("notes/d.md", 1, 1),
    ]
    assert run.progress[-1].total == run.summary.added + run.summary.updated


def test_two_changed_files_count_down_to_zero(
    db: Session, tmp_path: Path
) -> None:
    """The count the owner asked for, at the scale the label is read at.

    Two changed files among four: the denominator is 2 from the first report,
    the position advances only at the predicted files (the unchanged walk
    between them holds it), and ``total - done`` reaches 0 exactly when the
    last predicted file starts — where phase 193's ``N ≥ 1`` guard hides the
    count part. The terminal reports the same two files as updated."""
    root = _root(tmp_path, f"{PREFIX}two", _FOUR_FILES)
    _run(db, root)
    (root / "a.md").write_text("# A\n\nalpha edited\n", encoding="utf-8")
    (root / "notes" / "d.md").write_text(
        "# D\n\ndelta edited\n", encoding="utf-8"
    )

    run = _run(db, root)

    assert (run.summary.updated, run.summary.unchanged) == (2, 2)
    assert [(c.rel, c.done, c.total) for c in run.progress] == [
        ("a.md", 1, 2),
        ("b.md", 1, 2),  # unchanged: the walk moves, the count holds
        ("c.md", 1, 2),
        ("notes/d.md", 2, 2),
    ]
    remaining = [c.total - c.done for c in run.progress]
    assert remaining == [1, 1, 1, 0], (
        f"the label's remaining count went {remaining} — it must hold across "
        "the unchanged files and reach 0 at the last predicted one (where "
        "phase 193's guard hides the part)"
    )


# --------------------------------------------------------------------------
# Zero entity hydration
# --------------------------------------------------------------------------


def test_prescan_reads_one_column_and_hydrates_nothing(
    db: Session, tmp_path: Path, statement_capture: list[str]
) -> None:
    """The scan's statement projects EXACTLY ``documents.file_hash`` and the
    run's session is still empty when the walk begins.

    Three pins on the same claim, because the failure mode is the 2026-10-15
    incident (a full walk hydrating 252 MB of markdown):

    * every ``documents`` ``(source, path)`` SELECT issued before the first
      ``progress`` call — i.e. by the scan — names ``file_hash`` and NOTHING
      else, no ``content`` above all (the phase-185 column-list technique);
    * exactly one such read per importable file;
    * ``len(session.identity_map) == 0`` at the first ``progress`` call: the
      scan put no ``Document`` anywhere, so the run's retention stays bounded
      by the phase-184 cadence instead of growing with the KB.
    """
    root = _root(tmp_path, f"{PREFIX}hydrate", _FOUR_FILES)
    _run(db, root)
    db.expunge_all()
    statement_capture.clear()

    map_at_first_progress: list[int] = []

    def progress(source: str, rel: str, done: int, total: int) -> None:  # noqa: ARG001
        map_at_first_progress.append(len(db.identity_map))

    summary = asyncio.run(
        import_sources(
            [root], _llm(), session=db, progress=progress, on_scan_progress=None
        )
    )
    assert summary.unchanged == 4

    # The scan reads precede the loop's row loads in capture order, and the
    # first progress call is the boundary between them. Classified on the
    # PROJECTION (phase 185's technique): the scan's projection is the single
    # column token, the fast path's is every column but ``content``.
    def predicate(stmt: str) -> str:
        """Everything after the statement's ``FROM`` (case-insensitive —
        SQLAlchemy compiles the keyword upper-case)."""
        return re.split(r"\bfrom\s+documents\b", stmt, maxsplit=1, flags=re.I)[-1]

    doc_reads = [
        stmt
        for stmt in statement_capture
        if stmt.lstrip().lower().startswith("select")
        and re.search(r"\bfrom\s+documents\b", stmt, re.I)
        and "documents.source" in predicate(stmt)
        and "documents.path" in predicate(stmt)
    ]

    def projection(stmt: str) -> set[str]:
        head = re.split(r"\bfrom\b", stmt, maxsplit=1, flags=re.I)[0]
        return set(re.findall(r"documents\.([a-z_]+)", head))

    scan_reads = [s for s in doc_reads if projection(s) == {"file_hash"}]
    row_loads = [s for s in doc_reads if projection(s) != {"file_hash"}]
    assert len(scan_reads) == 4, (
        f"the unchanged run issued {len(scan_reads)} single-column file_hash "
        "reads — the scan reads each importable file's hash exactly once"
    )
    assert len(row_loads) == 4, (
        f"the loop issued {len(row_loads)} fast-path row loads — the scan must "
        "not replace _doc_row_stmt, which stays the only ROW load (phase 185)"
    )
    assert all(
        projection(s) == {"file_hash"} for s in scan_reads
    ), "a scan read projects more than file_hash — it would hydrate a row"
    assert all(
        "content" not in projection(s) for s in row_loads
    ), "the fast-path row load stopped deferring content (phase 185)"
    assert map_at_first_progress and map_at_first_progress[0] == 0, (
        "the run's session already held entities when the loop reached its "
        "first file — the pre-scan leaked a row into the identity map"
    )


# --------------------------------------------------------------------------
# The span is off the event loop (the phase-179 pattern)
# --------------------------------------------------------------------------


class _SlowHash:
    """The timing fixture: a PLAIN blocking ``_file_hash`` (like the real one
    — the point under test is WHERE it runs), counting its calls.

    The thread identity is NOT taken from here: ``_file_hash`` is called both
    by the scan (worker thread) and by the processing loop (loop thread), so
    the attribution comes from ``on_scan_progress``, which fires on the same
    worker thread as the scan's own hashes."""

    def __init__(self, real: Callable[[Path], str], block: float = HASH_BLOCK_S) -> None:
        self.real = real
        self.block = block
        self.calls = 0

    def __call__(self, path: Path) -> str:
        self.calls += 1
        time.sleep(self.block)
        return self.real(path)


def test_prescan_span_does_not_park_the_event_loop(
    db: Session, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    """THE phase-179 contract, re-pinned for the pre-scan.

    A 4-file scan with a 60 ms blocking hash is ~250 ms of work that used to
    be inline. The heartbeat coroutine polls every 5 ms; the ticks between two
    consecutive scan reports prove the LOOP kept running while the worker
    blocked (a regression that runs the scan inline parks the loop thread and
    the heartbeat cannot tick — the task-01 git measurement was 0 ticks over
    a 3 s clone). The thread identities are asserted too, so the pin fails on
    a machine fast enough to hide the latency.
    """
    root = _root(tmp_path, f"{PREFIX}freeze", _FOUR_FILES)
    stub = _SlowHash(importer._file_hash)
    monkeypatch.setattr(importer, "_file_hash", stub)

    ticks = 0
    scan_ticks: list[int] = []
    scan_positions: list[tuple[int, int]] = []
    scan_thread: list[int] = []

    def scan_hook(position: int, total: int) -> None:
        # Runs on the span's worker thread — the same thread that just hashed.
        scan_positions.append((position, total))
        scan_ticks.append(ticks)
        scan_thread.append(threading.get_ident())

    # A progress caller is what opens the gate (the CLI shape, ``progress=None``,
    # would skip the scan entirely) — this is the UI sync's shape.
    async def main() -> int:
        nonlocal ticks
        loop_ident = threading.get_ident()

        async def heartbeat() -> None:
            nonlocal ticks
            while True:
                await asyncio.sleep(0.005)
                ticks += 1

        hb = asyncio.create_task(heartbeat())
        try:
            await import_sources(
                [root],
                _llm(),
                session=db,
                progress=lambda s, r, d, t: None,
                on_scan_progress=scan_hook,
            )
        finally:
            hb.cancel()
        return loop_ident

    loop_ident = asyncio.run(main())

    assert scan_positions == [(1, 4), (2, 4), (3, 4), (4, 4)]
    assert scan_thread and loop_ident is not None
    assert all(ident != loop_ident for ident in scan_thread), (
        "the pre-scan ran on the event-loop thread — a long scan parks the "
        "single uvicorn loop again (phase 179's defect class)"
    )
    deltas = [
        b - a for a, b in zip([0, *scan_ticks], scan_ticks, strict=False)
    ]
    assert min(deltas) >= MIN_TICKS_PER_FILE, (
        f"the heartbeat ticked {deltas} times across consecutive scanned files "
        f"(expected >= {MIN_TICKS_PER_FILE} each) — the loop was starved while "
        "the scan blocked, so the span is not off-loop"
    )


# --------------------------------------------------------------------------
# The cancel (phase 134): the run stops, the worker finishes
# --------------------------------------------------------------------------


def test_cancel_during_the_prescan_stops_the_run_while_the_worker_finishes(
    db: Session, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    """The phase-134 cooperative cancel, measured at the scan's await.

    Cancelling the run ~80 ms into a ~250 ms scan raises ``CancelledError``
    out of ``import_sources`` PROMPTLY (well before the scan's own remaining
    work), NOTHING is indexed, and the worker thread runs to completion —
    threads are not interruptible, and this is exactly why the span reads
    through a session of its own instead of the run's (which the cancel
    closes on its way out). The pre-phase-194 pipeline had no await here at
    all, so this is the new boundary the UI's Stop button must survive.
    """
    root = _root(tmp_path, f"{PREFIX}cancel", _FOUR_FILES)
    stub = _SlowHash(importer._file_hash, block=0.12)
    monkeypatch.setattr(importer, "_file_hash", stub)

    real_prescan = importer._prescan_pending
    worker_finished = threading.Event()
    worker_results: list[Any] = []

    def spy(*args: Any, **kwargs: Any) -> Any:
        try:
            result = real_prescan(*args, **kwargs)
            worker_results.append(result)
            return result
        finally:
            worker_finished.set()

    monkeypatch.setattr(importer, "_prescan_pending", spy)

    async def main() -> float:
        # The hook runs on the WORKER thread, so it only appends to a list —
        # the house hook contract (a GIL-atomic store). Touching an
        # ``asyncio.Event`` from here would be a cross-thread loop call.
        seen: list[tuple[int, int]] = []

        def scan_hook(position: int, total: int) -> None:
            seen.append((position, total))

        task = asyncio.create_task(
            import_sources(
                [root],
                _llm(),
                session=db,
                progress=lambda s, r, d, t: None,
                on_scan_progress=scan_hook,
            )
        )
        # Wait (from the loop) until the scan has reported its first file.
        deadline = time.monotonic() + 10.0
        while not seen and time.monotonic() < deadline:
            await asyncio.sleep(0.005)
        assert seen, "the pre-scan never reported a file"
        await asyncio.sleep(0.08)
        t0 = time.monotonic()
        task.cancel()
        with pytest.raises(asyncio.CancelledError):
            await task
        return time.monotonic() - t0

    cancel_latency = asyncio.run(main())

    # The cancel is prompt: the run did NOT wait for the remaining scan work.
    assert cancel_latency < 0.5, (
        f"the cancelled run took {cancel_latency:.3f}s to unblock — it is "
        "still waiting for the blocking scan worker"
    )
    # The worker is not interruptible: it finishes its own session's work.
    assert worker_finished.wait(10.0), "the pre-scan worker never completed"
    assert stub.calls == 4, (
        f"the cancelled run's scan worker hashed {stub.calls} of 4 files — "
        "the contract is that the worker runs to completion"
    )
    # And the pipeline that would have consumed the prediction never ran.
    assert db.scalar(
        select(select(Document).where(Document.source == root.name).exists())
    ) is False, "the cancelled run indexed documents — the import never started"