"""Integration: phase 194 (task 02) — the ``"scan"`` phase on the status surface.

The owner report that opened the phase (2026-10-19): *"it said '1216
remaining' when it only had like 10 files to sync"*. Task 01 moved the
count onto the pre-scan's PREDICTED change set; this file pins the
status half (design D3, owner-confirmed A2 — the pre-scan is a VISIBLE
stage, not an invisible pause):

* the phase machine is ``null`` → ``"scan"`` (set immediately before
  ``import_sources``, where the scan actually runs) → ``"import"``
  (flipped by the FIRST per-file ``progress`` call) → ``"overview"`` →
  ``"summaries"`` → the terminal clears ``phase``;
* ``scan_done`` / ``scan_total`` carry the scan's own position through
  that stage, reset with the run, and are KEPT at every terminal beside
  the final file counts (the phase-64 keep-final-values convention);
* the payload is ADDITIVE — the pre-phase key set plus exactly those two
  keys, byte-identical in name and type, stable from idle through the
  terminal;
* the phase-187 sampling discipline holds on the new stage: the ``"scan"``
  set and the ``scan``→``import`` flip are TRANSITIONS and force the
  ``/proc`` read, while the scan's own per-file hook is THROTTLED — one
  read per interval, never one per file (LOCKED A4);
* the two degenerate/terminal edges: an empty knowledge base sits in
  ``"scan"`` and never reaches ``"import"`` (no ``progress`` call exists
  to flip it) and a cancel DURING the scan lands the ``cancelled``
  terminal with the scan counts kept and nothing indexed (the phase-134
  contract, the app-side twin of the importer-side pin in
  ``tests/integration/test_sync_pending_count.py``).

**Why one arm is choreographed and one is real.** An in-process import
never awaits real I/O (the embedder is a fake), so the loop cannot answer a
status poll from inside it — a real run's ``"import"`` phase is therefore
NOT reliably observable, and the real arm pins what the terminal keeps
(both pairs, divergent on the incremental run). The flip itself gets a
deterministic pin from the choreographed double (the ``test_sync_api``
gated-import idiom), which drives the runner's OWN closures and parks at
every frame the test reads. The scan, by contrast, runs in a
``to_thread`` worker, so the loop is FREE through it — that is exactly why
its frames are observable, and why the blocking-hash fixture slows ONLY the
scan window (a blanket stub would park the loop through the import too).

House patterns (``test_sync_api.py`` / ``test_sync_git_freeze.py``): the
``sync_client`` fixture is a CONTEXT-MANAGED ``TestClient`` — one app event
loop across requests, so the background run survives between the POST and
the polls; the module ``_status`` / ``_task`` / ``_cancel_event`` /
``_rss_sampled_at`` and the ``git_sources`` / ``sources_meta`` /
``folder_summaries`` / ``documents`` global state are reset around every
test; the probe is stubbed so nothing reaches the network, and the import
runs for real over a temp ``kind=local`` source with the deterministic
``FakeEmbedder``. The scan is made observable with a PLAIN blocking
``_file_hash`` stub scoped to the scan window (the phase-179 timing-fixture
idiom): the point under test is WHICH phase the run is in WHEN the page
polls, and a scan of instant hashes is over before the first poll.
"""
from __future__ import annotations

import asyncio
import logging
import threading
import time
from collections.abc import Callable, Iterator
from datetime import datetime
from pathlib import Path
from typing import Any

import pytest
from fastapi.testclient import TestClient
from sqlalchemy import select, text
from sqlalchemy.orm import Session

import app.rag.importer as importer
from app.api import sync as sync_api
from app.config import Settings
from app.db import SessionLocal, db_available
from app.main import app as fastapi_app
from app.models import Document, GitSource
from app.rag import git_sources as git_sources_resolver
from app.rag.importer import ImportSummary
from app.rag.llm import LLMClient
from tests.conftest import ADMIN_PASSWORD
from tests.fakes import FakeEmbedder

#: The probe's own source-name prefix — nothing else in the KB uses it,
#: so the module's DELETE-by-source cleanup can never touch another
#: suite's rows.
PREFIX = "phase194-t02-"

#: One blocking ``_file_hash`` per file. Three files is a ~360 ms scan:
#: long enough that a 10 ms poll catches several distinct positions inside
#: the phase (the frames are what the UI's 2 s poll would render), short
#: enough that the suite stays fast.
HASH_BLOCK_S = 0.12


# --- fixtures (the house sync hygiene) --------------------------------------


@pytest.fixture(autouse=True)
def _fresh_sync_state() -> Iterator[None]:
    """The module-level status object + task + cancel event + sampling
    guard are process-global — reset before AND after every test so no
    terminal state, stale cancel or stale throttle window leaks into the
    next one (``test_sync_api`` / ``test_sync_git_freeze``)."""
    sync_api._status = sync_api.SyncStatus()
    sync_api._task = None
    sync_api._cancel_event = asyncio.Event()
    sync_api._rss_sampled_at = float("-inf")
    yield
    sync_api._status = sync_api.SyncStatus()
    sync_api._task = None
    sync_api._cancel_event = asyncio.Event()
    sync_api._rss_sampled_at = float("-inf")


@pytest.fixture(autouse=True)
def clean_git_sources(db: Session) -> Iterator[None]:
    """The runner resolves through the real ``git_sources`` table —
    truncated around every test (the rows this file seeds are
    ``kind=local``)."""
    db.execute(text("TRUNCATE git_sources"))
    db.commit()
    yield
    db.execute(text("TRUNCATE git_sources"))
    db.commit()


@pytest.fixture(autouse=True)
def _reset_sources_version() -> Iterator[None]:
    """The phase-53 bump is global mutable state — pin it to the
    migration-0010 seed (0) around every test (own session: the runner
    bumps through its own short-lived ``SessionLocal``)."""
    if not db_available():
        pytest.skip("Postgres not reachable — run `podman compose up -d db` first")
    session = SessionLocal()
    try:
        session.execute(text("UPDATE sources_meta SET version = 0 WHERE id = 1"))
        session.commit()
        yield
    finally:
        session.execute(text("UPDATE sources_meta SET version = 0 WHERE id = 1"))
        session.commit()
        session.close()


@pytest.fixture(autouse=True)
def _clean_folder_summaries(db: Session) -> Iterator[None]:
    """``folder_summaries`` is global state the real generator writes on a
    changed KB — truncated around every test."""
    db.execute(text("TRUNCATE folder_summaries"))
    db.commit()
    yield
    db.execute(text("TRUNCATE folder_summaries"))
    db.commit()


@pytest.fixture(autouse=True)
def _clean_probe_documents(db: Session) -> Iterator[None]:
    """The real imports of this module write ``documents``/``chunks`` —
    removed before and after every test, narrowed to this module's own
    source names (the ``test_sync_pending_count`` hygiene)."""
    _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()


@pytest.fixture()
def sync_client() -> Iterator[TestClient]:
    """Context-managed ``TestClient`` — one app event loop across requests
    (the background run must survive between the POST and the polls)."""
    with TestClient(fastapi_app) as client:
        yield client


# --- helpers ----------------------------------------------------------------


def _settings(sources_dir: str) -> Settings:
    return Settings(_env_file=None, sources_dir=sources_dir)  # pyright: ignore[reportCallIssue]


def _stub_env(monkeypatch: pytest.MonkeyPatch) -> None:
    """Empty env fallback (the dev ``.env`` never leaks in) — this file's
    runs resolve their one local row from the table."""
    monkeypatch.setattr(
        git_sources_resolver,
        "get_settings",
        lambda: Settings(_env_file=None, git_sources=""),  # pyright: ignore[reportCallIssue]
    )


def _seed_local(db: Session, path: Path) -> None:
    """A ``kind=local`` row exactly as the phase-38 API stores it."""
    db.add(GitSource(url=str(path), kind="local", path=str(path)))
    db.commit()


def _local_source(tmp_path: Path, name: str, files: dict[str, str]) -> Path:
    """One source root (always created — the empty-KB arm needs a real
    empty directory, not a missing one), with *files* under it."""
    assert name.startswith(PREFIX), f"the probe source must be {PREFIX}-prefixed"
    root = tmp_path / name
    root.mkdir(parents=True, exist_ok=True)
    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 _login(client: TestClient) -> None:
    r = client.post("/api/login", json={"password": ADMIN_PASSWORD})
    assert r.status_code == 204, f"admin login failed: {r.status_code} {r.text}"


def _poll(client: TestClient, want: str, timeout: float = 20.0) -> dict:
    """Poll until ``state == want``, failing loudly on any other state."""
    deadline = time.monotonic() + timeout
    body: dict = {}
    while time.monotonic() < deadline:
        body = client.get("/api/sync/status").json()
        if body["state"] == want:
            return body
        assert body["state"] == "running", (
            f"unexpected state {body['state']!r} while waiting for {want!r}: {body}"
        )
        time.sleep(0.02)
    raise AssertionError(f"sync did not reach {want!r} within {timeout}s: {body}")


def _watch_scan(client: TestClient, timeout: float = 20.0) -> tuple[list[dict], dict | None]:
    """Poll through the run's SCAN window; stop at the first frame that is
    no longer a scan.

    Returns ``(frames, terminal)`` — ``terminal`` is non-``None`` when the
    run finished inside the window. Both stops are honest: the scan runs
    OFF the loop (so its frames are served, and this list is exactly what
    the label was built from), while the import that follows is one
    straight run of the importer that can finish before the next poll.
    The flip frame therefore gets pinned deterministically by the
    choreographed test below; what the real run pins here is the scan's
    own frames plus the values the terminal keeps."""
    frames: list[dict] = []
    terminal: dict | None = None
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        body = client.get("/api/sync/status").json()
        if body["state"] != "running":
            terminal = body
            break
        frames.append(body)
        if body["phase"] != "scan":
            break
        time.sleep(0.01)
    else:
        raise AssertionError(f"the run never left the scan phase: {frames[-1]}")
    return frames, terminal


class _FakeScanImport:
    """A parked ``import_sources`` double for the app-side edges.

    It records the two position closures the runner passes (the wiring
    pin) and fires NOTHING — which is exactly what a run with no
    importable file does (no scan report, so no position, and no
    ``progress`` call to flip the phase; the importer-side proof lives in
    the empty-KB arm below and in ``test_sync_pending_count.py``). The
    park lets the test read the status inside the ``"scan"`` phase without
    racing the pipeline; the caller always releases it in ``finally``.
    """

    def __init__(self, summary: ImportSummary | None = None) -> None:
        self.summary = summary if summary is not None else ImportSummary()
        self.started = threading.Event()
        self.release = threading.Event()
        self.progress_hook: Callable[[str, str, int, int], None] | None = None
        self.scan_hook: Callable[[int, int], None] | None = None

    async def __call__(
        self,
        sources: list[Path],
        llm: LLMClient,
        *,
        prune: bool = False,
        limit: int | None = None,
        session: Session | None = None,
        progress: Callable[[str, str, int, int], None] | None = None,
        on_chunk_progress: Callable[[int, int], None] | None = None,
        on_image_progress: Callable[[int, int], None] | None = None,
        on_scan_progress: Callable[[int, int], None] | None = None,
        ignore_by_root: dict[str, list[str]] | None = None,
        include_hidden_by_root: dict[str, bool] | None = None,
        doc_dates_by_root: dict[str, dict[str, datetime]] | None = None,
    ) -> ImportSummary:
        self.progress_hook = progress
        self.scan_hook = on_scan_progress
        self.started.set()
        await asyncio.to_thread(self.release.wait)
        return self.summary


def _install_counting_memory(monkeypatch: pytest.MonkeyPatch) -> list[dict[str, Any]]:
    """Replace the two memory helpers on ``app.api.sync`` with recorders.

    ``_sample_rss`` calls them ONLY when the throttle lets it, so one
    entry per entry below is one real read — and each entry carries the
    phase (and the scan/file position) at that instant, which is the
    difference between "a transition forced a read" and "a per-file hook
    read once per file" (LOCKED A4)."""
    reads: list[dict[str, Any]] = []

    def _record() -> None:
        reads.append(
            {
                "phase": sync_api._status.phase,
                "scan_done": sync_api._status.scan_done,
                "files_done": sync_api._status.files_done,
                "current_file": sync_api._status.current_file,
            }
        )

    def fake_rss_mb() -> int:
        _record()
        return 100

    def fake_rss_peak_mb() -> int:
        return 120

    monkeypatch.setattr(sync_api, "rss_mb", fake_rss_mb)
    monkeypatch.setattr(sync_api, "rss_peak_mb", fake_rss_peak_mb)
    return reads


def _slow_hash(
    monkeypatch: pytest.MonkeyPatch, block: float = HASH_BLOCK_S
) -> dict[str, bool]:
    """Make ONLY the pre-scan window slow, and return its flag.

    ``_file_hash`` is called by the scan AND by the processing loop, so a
    blanket blocking stub would also park the event loop through the whole
    import (an in-process import never awaits real I/O, so the loop cannot
    answer a status poll inside it) — the frames this file needs would
    then be unservable rather than absent. The flag is raised around
    :func:`app.rag.importer._prescan_pending` only, so the scan's window is
    the fixture's window: the loop stays free exactly while the run is in
    the phase under test (the phase-179 off-loop claim, observed from the
    HTTP side), and the import after it runs at full speed.

    The flag is a plain ``dict`` entry written/read from two threads: the
    house hook contract (a GIL-atomic store) applies here for the same
    reason it applies to the runner's closures.
    """
    real_hash = importer._file_hash
    real_prescan = importer._prescan_pending
    state = {"in_scan": False}

    def slow(path: Path) -> str:
        if state["in_scan"]:
            time.sleep(block)
        return real_hash(path)

    def spy_prescan(*args: Any, **kwargs: Any) -> Any:
        state["in_scan"] = True
        try:
            return real_prescan(*args, **kwargs)
        finally:
            state["in_scan"] = False

    monkeypatch.setattr(importer, "_file_hash", slow)
    monkeypatch.setattr(importer, "_prescan_pending", spy_prescan)
    return state


class _ChoreographedImport:
    """An ``import_sources`` double that replays the importer's hook script
    through the runner's OWN closures, parking at every point the test
    reads the status.

    The choreography (all four stops are ``threading.Event`` pairs, so no
    test here races the pipeline):

    1. fire ``scan_pairs[:1]`` through ``on_scan_progress`` → ``first``
       (the test reads the scan's FIRST position);
    2. on ``go_on``: fire the rest → ``scanned`` (the test reads the last
       position, still inside the phase);
    3. on ``go_import``: fire ``progress_call`` through ``progress`` —
       THE FLIP → ``flipped`` (the test reads the ``"import"`` frame);
    4. on ``release``: return the canned summary.
    """

    def __init__(
        self,
        summary: ImportSummary,
        scan_pairs: list[tuple[int, int]],
        progress_call: tuple[str, str, int, int],
    ) -> None:
        self.summary = summary
        self.scan_pairs = list(scan_pairs)
        self.progress_call = progress_call
        self.first = threading.Event()
        self.go_on = threading.Event()
        self.scanned = threading.Event()
        self.go_import = threading.Event()
        self.flipped = threading.Event()
        self.release = threading.Event()
        self.scan_hook: Callable[[int, int], None] | None = None
        self.progress_hook: Callable[[str, str, int, int], None] | None = None

    async def __call__(
        self,
        sources: list[Path],
        llm: LLMClient,
        *,
        prune: bool = False,
        limit: int | None = None,
        session: Session | None = None,
        progress: Callable[[str, str, int, int], None] | None = None,
        on_chunk_progress: Callable[[int, int], None] | None = None,
        on_image_progress: Callable[[int, int], None] | None = None,
        on_scan_progress: Callable[[int, int], None] | None = None,
        ignore_by_root: dict[str, list[str]] | None = None,
        include_hidden_by_root: dict[str, bool] | None = None,
        doc_dates_by_root: dict[str, dict[str, datetime]] | None = None,
    ) -> ImportSummary:
        # The wiring pin: both closures arrive live (None here would mean
        # the runner stopped passing them).
        self.scan_hook = on_scan_progress
        self.progress_hook = progress
        assert on_scan_progress is not None and progress is not None
        for pair in self.scan_pairs[:1]:
            on_scan_progress(*pair)
        self.first.set()
        await asyncio.to_thread(self.go_on.wait)
        for pair in self.scan_pairs[1:]:
            on_scan_progress(*pair)
        self.scanned.set()
        await asyncio.to_thread(self.go_import.wait)
        progress(*self.progress_call)
        self.flipped.set()
        await asyncio.to_thread(self.release.wait)
        return self.summary


def _stub_probe(monkeypatch: pytest.MonkeyPatch) -> None:
    async def fake_check_models(llm: LLMClient) -> None:
        return None

    monkeypatch.setattr(sync_api, "check_models", fake_check_models)


def _install_local_run(
    monkeypatch: pytest.MonkeyPatch, db: Session, tmp_path: Path, source: Path
) -> None:
    """Point one app run at *source*: local row, tmp sources root, stubbed
    probe, the deterministic in-process embedder (the real import, no
    network)."""
    _seed_local(db, source)
    _stub_env(monkeypatch)
    monkeypatch.setattr(
        sync_api, "get_settings", lambda: _settings(sources_dir=str(tmp_path / "bor"))
    )
    _stub_probe(monkeypatch)
    monkeypatch.setattr(sync_api, "LLMClient", lambda: FakeEmbedder())


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


def _log_lines(caplog: pytest.LogCaptureFixture, prefix: str) -> list[str]:
    return [r.getMessage() for r in caplog.records if r.getMessage().startswith(prefix)]


# --- the phase-machine sequence --------------------------------------------


def test_real_run_reports_the_scan_frames_and_keeps_both_pairs(
    sync_client: TestClient,
    monkeypatch: pytest.MonkeyPatch,
    db: Session,
    tmp_path: Path,
) -> None:
    """D3 through the REAL pipeline, twice (one changed file included).

    Run 1 (a first import): the prelude reports ``phase: null`` with
    ``scan_*`` 0/0, then the run reports ``"scan"`` with ``scan_total`` at
    the importable count and ``scan_done`` climbing through AT LEAST two
    distinct positions (the frames the ``Scanning… (n/total)`` label
    needs) with ``current_file`` null throughout — the scan names no file.
    The terminal clears ``phase`` and KEEPS the final scan pair beside the
    file pair, and on a first import both pairs are 3/3 (A1's degenerate
    parity: walk == work, the only shape the phase-193 suite ever saw).

    Run 2 (one file edited — the owner's prod shape at 1-of-3): the scan
    denominator is STILL the importable 3, while the kept file pair is
    ``1/1`` and the terminal reports exactly one updated document. That
    divergence — the scan counted the walk, the count counted the work —
    IS the phase, and it survives on the terminal where no poll can miss
    it."""
    source = _local_source(tmp_path, f"{PREFIX}machine", _THREE_FILES)
    _install_local_run(monkeypatch, db, tmp_path, source)
    _slow_hash(monkeypatch)

    _login(sync_client)
    idle = sync_client.get("/api/sync/status").json()
    assert idle["phase"] is None
    assert idle["scan_done"] == 0 and idle["scan_total"] == 0

    assert sync_client.post("/api/sync").status_code == 202
    frames, _late = _watch_scan(sync_client)
    scan_frames = [f for f in frames if f["phase"] == "scan"]
    assert scan_frames, f"the run never reported the scan phase: {frames}"
    assert all(f["current_file"] is None for f in scan_frames), (
        "the scan named a file — current_file stays null until the first "
        "progress call flips the phase to import"
    )
    # ``0`` is the transition frame — the phase is set before the scan
    # reports its first file, and the UI's total===0 guard renders it bare.
    totals = {f["scan_total"] for f in scan_frames}
    assert totals <= {0, 3} and max(totals) == 3, totals
    positions = sorted({f["scan_done"] for f in scan_frames if f["scan_total"] == 3})
    assert len(positions) >= 2, (
        f"only scan position(s) {positions} were observable — the label needs "
        "the scan's own n/total to move (the fixture scan blocks on purpose)"
    )
    assert positions[0] >= 1 and positions[-1] <= 3

    body = _poll(sync_client, "success")
    assert body["detail"]["added"] == 3
    assert body["phase"] is None and body["current_file"] is None
    # The keep-final-values convention, beside the file counts.
    assert body["scan_total"] == 3 and body["scan_done"] == 3
    assert body["files_total"] == 3 and body["files_done"] == 3

    # Run 2 — one changed file of three (the owner's prod case, scaled).
    (source / "b.md").write_text("# B\n\nbravo content EDITED\n", encoding="utf-8")
    assert sync_client.post("/api/sync").status_code == 202
    frames2, _late2 = _watch_scan(sync_client)
    scan2 = [f for f in frames2 if f["phase"] == "scan"]
    totals2 = {f["scan_total"] for f in scan2 if f["scan_total"]}
    assert scan2 and totals2 == {3}, (
        "the scan denominator is the importable count, not the prediction"
    )
    assert sorted({f["scan_done"] for f in scan2 if f["scan_total"] == 3})[-1] >= 2

    body2 = _poll(sync_client, "success")
    assert (body2["detail"]["added"], body2["detail"]["updated"]) == (0, 1)
    assert body2["phase"] is None
    # The divergence the phase exists for: the kept scan pair is the WALK
    # it did (3 files hashed), the kept file pair is the WORK (1 file).
    assert (body2["scan_done"], body2["scan_total"]) == (3, 3)
    assert (body2["files_done"], body2["files_total"]) == (1, 1)


def test_scan_flips_to_import_on_the_first_progress_call(
    sync_client: TestClient,
    monkeypatch: pytest.MonkeyPatch,
    db: Session,
    tmp_path: Path,
) -> None:
    """The sequence, frame by frame, with the flip pinned.

    The in-process import runs straight through without awaiting real I/O,
    so a poll cannot reliably land inside the ``"import"`` phase (that is
    why the real arm above pins the terminal's kept pairs). This arm drives
    the runner's OWN closures through a choreographed double — the
    ``test_sync_api`` gated-import idiom — and reads the status at each
    stop: prelude ``null``/0/0 → ``"scan"`` at position 1 of 3 → still
    ``"scan"`` at position 3 of 3 → ``"import"`` with the PREDICTED
    ``files_total`` (1, the count that reaches ``N remaining``) and the
    scan pair still showing 3/3 → the terminal, which clears ``phase`` and
    keeps both pairs."""
    source = _local_source(tmp_path, f"{PREFIX}flip", _THREE_FILES)
    _install_local_run(monkeypatch, db, tmp_path, source)
    gated = _ChoreographedImport(
        ImportSummary(files=3, updated=1, unchanged=2),
        scan_pairs=[(1, 3), (2, 3), (3, 3)],
        progress_call=(source.name, "b.md", 1, 1),
    )
    monkeypatch.setattr(sync_api, "import_sources", gated)

    _login(sync_client)
    idle = sync_client.get("/api/sync/status").json()
    assert idle["state"] == "idle"
    assert idle["phase"] is None
    assert idle["scan_done"] == 0 and idle["scan_total"] == 0

    assert sync_client.post("/api/sync").status_code == 202
    try:
        assert gated.first.wait(10.0), "the run never reached the scan"
        s1 = sync_client.get("/api/sync/status").json()
        assert s1["state"] == "running"
        assert s1["phase"] == "scan"
        assert (s1["scan_done"], s1["scan_total"]) == (1, 3)
        assert s1["current_file"] is None
        # Still the pre-import phase: no file count yet.
        assert (s1["files_done"], s1["files_total"]) == (0, 0)

        gated.go_on.set()
        assert gated.scanned.wait(10.0), "the scan never reported its last file"
        s2 = sync_client.get("/api/sync/status").json()
        assert s2["phase"] == "scan"
        assert (s2["scan_done"], s2["scan_total"]) == (3, 3)

        gated.go_import.set()
        assert gated.flipped.wait(10.0), "the progress hook never fired"
        s3 = sync_client.get("/api/sync/status").json()
        assert s3["phase"] == "import", (
            "the first progress call must flip scan -> import: "
            f"{ {k: s3[k] for k in ('phase', 'current_file', 'files_done', 'files_total')} }"
        )
        assert s3["current_file"] == f"{source.name}/b.md"
        # The flip already carries the PREDICTED denominator — the number
        # the label turns into "N remaining".
        assert (s3["files_done"], s3["files_total"]) == (1, 1)
        # … and the scan pair stays where the scan left it.
        assert (s3["scan_done"], s3["scan_total"]) == (3, 3)
    finally:
        gated.go_on.set()
        gated.go_import.set()
        gated.release.set()

    body = _poll(sync_client, "success")
    assert body["phase"] is None and body["current_file"] is None
    assert (body["scan_done"], body["scan_total"]) == (3, 3)
    assert (body["files_done"], body["files_total"]) == (1, 1)


# --- the additive payload ----------------------------------------------------


def test_scan_keys_are_additive_and_stable_from_idle_to_terminal(
    sync_client: TestClient,
    monkeypatch: pytest.MonkeyPatch,
    db: Session,
    tmp_path: Path,
) -> None:
    """The additive-only contract on the wire (the phase-187 discipline).

    The payload's key set is EXACTLY the pre-phase-194 set plus
    ``scan_done`` / ``scan_total``; it does not change between idle,
    mid-run and the terminal (no key appears or disappears with the
    phase); and both new keys are plain ``int`` — 0 when there is nothing
    to report, never ``null`` and never a plausible position."""
    expected_pre = {
        "state",
        "started_at",
        "finished_at",
        "detail",
        "error",
        "current_file",
        "files_done",
        "files_total",
        "current_chunk",
        "chunks_total",
        "images_done",
        "images_total",
        "phase",
        "current_summary",
        "summaries_done",
        "summaries_total",
        "rss_mb",
        "rss_peak_mb",
    }
    source = _local_source(tmp_path, f"{PREFIX}keys", _THREE_FILES)
    _install_local_run(monkeypatch, db, tmp_path, source)
    _stub_probe(monkeypatch)

    _login(sync_client)
    idle = sync_client.get("/api/sync/status").json()
    assert set(idle) == expected_pre | {"scan_done", "scan_total"}, set(idle)
    assert idle["scan_done"] == 0 and idle["scan_total"] == 0

    assert sync_client.post("/api/sync").status_code == 202
    mid = sync_client.get("/api/sync/status").json()
    assert set(mid) == expected_pre | {"scan_done", "scan_total"}
    assert mid["state"] == "running"

    body = _poll(sync_client, "success")
    assert set(body) == expected_pre | {"scan_done", "scan_total"}
    for key in ("scan_done", "scan_total"):
        assert isinstance(body[key], int) and not isinstance(body[key], bool), body
    # Every pre-existing key keeps its type too (the additive pin's other
    # half — the file pair is still ints, the phase still a string|null).
    assert isinstance(body["files_done"], int) and isinstance(body["files_total"], int)
    assert body["phase"] is None


# --- the sampling discipline ------------------------------------------------


def test_scan_and_import_transitions_force_the_sample_and_the_scan_hook_does_not(
    sync_client: TestClient,
    monkeypatch: pytest.MonkeyPatch,
    db: Session,
    tmp_path: Path,
) -> None:
    """Phase 187's rule, extended to the new stage (LOCKED A4).

    Every entry below is one REAL ``/proc`` read (the two helpers are only
    called when the throttle lets ``_sample_rss`` through), tagged with
    the phase it happened in. Collapsing consecutive same-phase reads, the
    sequence is one read per transition: run start (``phase`` still null)
    → ``"scan"`` → ``"import"`` (the flip) → ``"overview"`` →
    ``"summaries"`` → the terminal (null again). The scan transition's
    read carries ``scan_done == 0`` — it is the transition, not a file —
    and there is EXACTLY one read inside the whole scan phase, so a
    per-file scan read (one ``/proc`` read per file, the regression this
    pins) fails here even though the fixture scanned three files."""
    source = _local_source(tmp_path, f"{PREFIX}sampling", _THREE_FILES)
    _install_local_run(monkeypatch, db, tmp_path, source)
    _slow_hash(monkeypatch)
    reads = _install_counting_memory(monkeypatch)

    _login(sync_client)
    assert sync_client.post("/api/sync").status_code == 202
    body = _poll(sync_client, "success")

    assert reads, "the run sampled no memory at all"
    phases = [r["phase"] for r in reads]
    transitions = [p for i, p in enumerate(phases) if i == 0 or p != phases[i - 1]]
    assert transitions == [None, "scan", "import", "overview", "summaries", None], reads

    scan_reads = [r for r in reads if r["phase"] == "scan"]
    assert len(scan_reads) == 1, (
        f"the scan phase cost {len(scan_reads)} /proc reads for 3 files — the "
        "per-file scan hook must sample UNFORCED (one read per interval)"
    )
    assert scan_reads[0]["scan_done"] == 0, (
        "the scan-phase read is not the transition — the forced sample must "
        "land when the phase is SET, before the first file is scanned"
    )
    flip_reads = [r for r in reads if r["phase"] == "import"]
    assert flip_reads and flip_reads[0]["current_file"] is not None, (
        "no read at the scan->import flip — the transition is a forced "
        f"sampling point: {reads}"
    )
    # The payload reports what the sampler wrote, not a stale pair.
    assert (body["rss_mb"], body["rss_peak_mb"]) == (100, 120), body


# --- the degenerate empty knowledge base ------------------------------------


def test_empty_kb_stays_in_scan_and_never_reaches_import(
    sync_client: TestClient,
    monkeypatch: pytest.MonkeyPatch,
    db: Session,
    tmp_path: Path,
) -> None:
    """The D1 degenerate clause, pinned on BOTH sides.

    Importer side (real): an empty source dir makes the pre-scan report
    NOTHING — no ``on_scan_progress`` call, no ``progress`` call, a zero
    denominator. App side (the double that fires nothing, i.e. exactly
    what that run reports): the status sits in ``"scan"`` — the flip
    lives in the per-file hook and no hook ever fires — with ``scan_*``
    0/0 and ``current_file`` null, and the terminal clears ``phase``
    while keeping the 0/0 pair (the UI's ``total === 0`` guard then
    renders the bare label with no count part)."""
    # Importer side: the real scan of an empty root reports nothing.
    empty = _local_source(tmp_path, f"{PREFIX}empty", {})
    llm = FakeEmbedder()
    scan_calls: list[tuple[int, int]] = []
    progress_calls: list[tuple[str, str, int, int]] = []
    summary = asyncio.run(
        importer.import_sources(
            [empty],
            llm,
            progress=lambda s, r, d, t: progress_calls.append((s, r, d, t)),
            on_scan_progress=lambda p, t: scan_calls.append((p, t)),
        )
    )
    assert (scan_calls, progress_calls) == ([], [])
    assert summary.files == 0

    # App side: a run that reports nothing is a run that stays "scan".
    _install_local_run(monkeypatch, db, tmp_path, empty)
    fake_import = _FakeScanImport()
    monkeypatch.setattr(sync_api, "import_sources", fake_import)

    _login(sync_client)
    assert sync_client.post("/api/sync").status_code == 202
    try:
        assert fake_import.started.wait(10.0), "the run never reached the import"
        # Wiring: the runner passed BOTH live closures (so the empty
        # phase above is the scan reporting nothing, not a missing hook).
        assert fake_import.scan_hook is not None
        assert fake_import.progress_hook is not None
        body = sync_client.get("/api/sync/status").json()
        assert body["state"] == "running"
        assert body["phase"] == "scan"
        assert body["current_file"] is None
        assert body["scan_done"] == 0 and body["scan_total"] == 0
        assert body["files_done"] == 0 and body["files_total"] == 0
    finally:
        fake_import.release.set()

    terminal = _poll(sync_client, "success")
    assert terminal["phase"] is None
    assert terminal["scan_done"] == 0 and terminal["scan_total"] == 0
    assert terminal["files_done"] == 0 and terminal["files_total"] == 0


# --- the cancel during the pre-scan -----------------------------------------


def test_cancel_during_the_scan_lands_cancelled_with_the_scan_counts_kept(
    sync_client: TestClient,
    monkeypatch: pytest.MonkeyPatch,
    db: Session,
    tmp_path: Path,
    caplog: pytest.LogCaptureFixture,
) -> None:
    """The phase-134 contract at the pipeline's NEW await (the app-side
    twin of the importer-side cancel pin).

    The scan is blocking on purpose, so the cancel provably lands inside
    it: ``POST /api/sync/cancel`` answers, the ``cancelled`` terminal
    arrives well before the abandoned scan's remaining work could finish,
    ``phase`` clears, the position the scan had reached is KEPT beside the
    (still zero) file pair, ``error`` stays null and ``detail`` empty (a
    cancel is intent, not failure), and NOTHING was indexed — the walk
    never started. The lifecycle line keeps its exact pre-phase prefix
    (LOCKED A2's append-only rule)."""
    source = _local_source(tmp_path, f"{PREFIX}cancel", _THREE_FILES)
    _install_local_run(monkeypatch, db, tmp_path, source)
    _slow_hash(monkeypatch, block=0.4)

    _login(sync_client)
    with caplog.at_level(logging.INFO, logger="app.api.sync"):
        assert sync_client.post("/api/sync").status_code == 202
        # Wait for proof the run is INSIDE the scan: the phase, and a
        # position that moved.
        deadline = time.monotonic() + 20.0
        live: dict = {}
        while time.monotonic() < deadline:
            live = sync_client.get("/api/sync/status").json()
            if live["phase"] == "scan" and live["scan_done"] >= 1:
                break
            assert live["state"] == "running", live
            time.sleep(0.01)
        assert live.get("phase") == "scan" and live["scan_done"] >= 1, (
            f"the cancel never landed inside the scan: {live}"
        )
        position_at_cancel = live["scan_done"]

        t0 = time.monotonic()
        r = sync_client.post("/api/sync/cancel")
        assert r.status_code == 202, f"cancel not accepted: {r.status_code} {r.text}"
        assert time.monotonic() - t0 < 2.0, "the cancel request itself stalled"

        body = _poll(sync_client, "cancelled", timeout=3.0)

    assert body["error"] is None
    assert body["detail"] == {}
    assert body["phase"] is None
    assert body["current_file"] is None
    # The scan's final position is kept (the keep-final-values
    # convention), and it is the position the cancel found — the run did
    # not finish the scan it was cancelled in.
    assert body["scan_total"] == 3
    assert body["scan_done"] >= position_at_cancel
    assert body["files_done"] == 0 and body["files_total"] == 0

    # Nothing was indexed: the walk never started.
    assert db.scalar(
        select(select(Document).where(Document.source == source.name).exists())
    ) is False

    # The lifecycle line keeps its pre-phase-194 prefix exactly.
    cancelled = _log_lines(caplog, "sync: cancelled by user state=cancelled files_done=")
    assert len(cancelled) == 1, cancelled


# --- the kill switch's status shape ----------------------------------------


def test_kill_switch_off_reports_no_scan_position_but_the_same_machine(
    sync_client: TestClient,
    monkeypatch: pytest.MonkeyPatch,
    db: Session,
    tmp_path: Path,
) -> None:
    """``BOR_SYNC_PENDING_COUNT=0`` on the STATUS side.

    The importer takes the phase-193 stats pre-walk, which reports NOTHING,
    so the scan pair stays 0/0 for the whole run (the page renders no
    count from it) while ``files_done`` / ``files_total`` carry the WALK
    position again — 3 walked, not 1 predicted. The phase machine itself is
    UNCHANGED by the switch (the pre-``import_sources`` set is still
    ``"scan"`` and the first file still flips it: that flip is pinned in
    the choreographed test above, and the switch's importer-side halves are
    pinned in ``tests/unit/test_sync_pending_count.py`` and
    ``tests/integration/test_sync_pending_count.py``). The pins below read
    the TERMINAL, which is where both pairs survive deterministically."""
    source = _local_source(tmp_path, f"{PREFIX}off", _THREE_FILES)
    _install_local_run(monkeypatch, db, tmp_path, source)
    # The switch lives on the settings object the import reads (llm).
    real_embedder = FakeEmbedder()
    real_embedder.settings = Settings(_env_file=None, sync_pending_count=0)  # pyright: ignore[reportCallIssue]
    monkeypatch.setattr(sync_api, "LLMClient", lambda: real_embedder)

    _login(sync_client)
    assert sync_client.post("/api/sync").status_code == 202
    body = _poll(sync_client, "success")

    assert body["detail"]["added"] == 3
    # No scan, so no position — kept at 0/0, the value the UI renders as
    # "no count part" rather than a stale or invented one.
    assert body["scan_done"] == 0 and body["scan_total"] == 0
    # The WALK count is back (phase-193 semantics, byte-for-byte).
    assert (body["files_done"], body["files_total"]) == (3, 3)
    assert body["phase"] is None