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

"""Integration: the admin sources-sync API (phase 32, task 01; phase 35,
task 03 re-points the URL resolution at the shared resolver; phase 38,
task 03 adds the local kind).

Covers the in-process sync runner end to end over HTTP: anonymous 403s
on both endpoints; admin idle → 202 → ``success`` with the full
ImportSummary detail; 409 on a double trigger while a run is in flight;
``GitSyncError`` → ``failed`` with the failing repo named and **zero**
import attempts; empty on *both* origins (no git rows, no local rows,
no env URLs) → ``failed`` loudly (``no sources configured
(git or local)``); an embedding failure → ``failed`` with any
credentials masked; the import always runs with ``prune=True``; and the
phase-31 overview trigger is change-gated (no ``lite`` call on an
unchanged KB).

Phase 38 (local kind): local-only, git-only, and mixed syncs over a
**host temp local dir** (the app server runs on the same host) — the
mixed run goes through the **real** ``import_sources`` (deterministic
in-process ``FakeEmbedder``, no network), so the local file verifiably
lands in the KB via ``GET /api/docs`` and union pruning holds (a file
deleted out of the local dir is pruned on the next sync while the git
doc survives); a local directory missing at sync time → ``failed`` with
``local source missing: <path>`` and **zero** import attempts.

Phase 35: the runner resolves the sources through
:func:`app.rag.git_sources.effective_sources` — the **real** resolver
against the **real** ``git_sources`` table (truncated around every
test), so DB-over-env and the env fallback go through the actual
indirection; the env list is driven by a fresh ``Settings`` on the
resolver's module (the dev ``.env`` never leaks in).

Phase 41: the pre-sync model probe (``check_models``) — a dead model
endpoint fails the run **before any clone** (the model-naming error
lands in the ``failed`` state verbatim; the sanitizer is a no-op for
it); the probe runs before ``effective_sources``; a real probe against
a stubbed dead ``LLMClient`` names the embed model **and** masks the
credentials wrapped from the endpoint URL. Where the runner keeps a
real ``LLMClient`` the probe is stubbed (:func:`_stub_probe`) so no
test ever hits the network; the ``_real_llm`` tests get a passing
probe from ``FakeEmbedder.embed_one``/``chat``.

Phase 53 (task 02): the sources-version bump — a sync whose import
changed the KB (added + updated + pruned > 0) advances the single-row
``sources_meta`` counter exactly once (the new generation lands in the
``/api/sync/status`` detail as ``sources_version``); an unchanged
re-sync never bumps (the detail still reports the current generation),
and every failure path (git error, model down) never bumps. The
counter is pinned to the migration-0010 seed (0) around every test by
:func:`_reset_sources_version`.

The git / import / overview / folder-summary layers are monkeypatched
in ``app.api.sync`` (same fake style as ``test_import_docs_git.py``) —
no real git, no LLM: the runner's state machine and HTTP surface are
under test. Phase 94 (task 02): the folder-summary layer gets the same
treatment — the fake-import tests' canned summaries would otherwise
steer the REAL ``generate_folder_summaries`` at the global
``documents`` table and the real ``LLMClient`` (network); the
real-import tests (host temp dirs, deterministic ``FakeEmbedder``) keep
the real generator, with ``folder_summaries`` truncated around every
test (:func:`_clean_folder_summaries`).

The admin client is used **as a context manager** on purpose: the
background sync task lives on the app's event loop, so the loop must
survive across requests — exactly how the app runs under uvicorn.
(A TestClient without the context manager starts a fresh loop per
request and would cancel the task on request exit.)

Phase 127 (task 03): the changed-KB branch passes the import's change
set to the generator (``changed_docs=summary.changed_paths`` —
incremental regeneration over the changed set's folder keys only),
the ``changed=N`` log line names the change count, and the status's
``summaries_total`` is the affected count (the hook's total) rather
than the table's candidate count. The unchanged-walk gap probe is
orthogonal: it still calls the generator exactly as before
(``only_missing=True``, no change set). The fakes above record the
``changed_docs=`` kwarg (``FakeFolderSummaries.changed_docs_sets``) so
the wiring is asserted, not inferred.

Phase 134 (task 01): the cancel — anonymous 403 on the new
``POST /api/sync/cancel``; admin: 409 ``no sync is running`` idle and
after a terminal run, 202 ``cancel requested`` while a run is in
flight (the run is parked inside the import — the file's existing
gated-import pattern over the app's own event loop); the status then
reports the new ``cancelled`` terminal (``error`` null, ``detail``
empty, the final file counts kept, the chunk position cleared) and a
second run after the cancel completes to ``success`` (the restart is
clean — the stale cancel event did not bleed in, and only the
completed run bumps the sources version).
"""
from __future__ import annotations

import asyncio
import logging
import os
import subprocess
import textwrap
import threading
import time
from collections.abc import Callable, Iterator, Sequence
from datetime import UTC, datetime
from pathlib import Path
from typing import Any

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

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.folder_summaries import affected_folder_keys
from app.rag.importer import ImportSummary
from app.rag.llm import EmbeddingError, LLMClient, ModelUnavailableError
from app.rag.sources_meta import current_sources_version
from scripts.git_sync import GitSyncError, clone_or_pull
from tests.conftest import ADMIN_PASSWORD
from tests.fakes import FakeEmbedder


@pytest.fixture(autouse=True)
def _fresh_sync_state() -> Iterator[None]:
    """The module-level status object + task are process-global: reset them
    around every test (both before — a previous test's terminal state
    would leak into the idle assertion — and after).

    Phase 134 (task 01): the module ``_cancel_event`` joins the reset —
    a stale (set) event from a previous test's cancelled run would
    otherwise bleed into the next run's boundary checks."""
    sync_api._status = sync_api.SyncStatus()
    sync_api._task = None
    sync_api._cancel_event = asyncio.Event()
    yield
    sync_api._status = sync_api.SyncStatus()
    sync_api._task = None
    sync_api._cancel_event = asyncio.Event()


@pytest.fixture(autouse=True)
def clean_git_sources(db: Session) -> Iterator[None]:
    """Phase 35: the runner resolves through the real ``git_sources``
    table — global state, truncated around every test (the ``db``
    fixture skips the file when Postgres is down)."""
    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]:
    """Phase 53: the sources version counter is global mutable state —
    pin it to the migration-0010 seed (0) around every sync test so the
    bump assertions start from a known generation (own session: the
    runner bumps through its own short-lived ``SessionLocal``).
    Skips like the ``db`` fixture when Postgres is down."""
    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()
def sync_client() -> Iterator[TestClient]:
    """Context-managed TestClient — one app event loop across requests
    (the background task must survive between the POST and the polls)."""
    with TestClient(fastapi_app) as client:
        yield client


def _settings(sources_dir: str = "~/bor-sources") -> Settings:
    """Fresh settings (no .env file); explicit kwargs beat any env leaks.

    (The ``git_sources`` kwarg is gone with phase 35 — the runner reads
    the URLs from the resolver, not from its own settings; the env list
    is stubbed on the resolver's module via :func:`_stub_env`.)
    """
    return Settings(_env_file=None, sources_dir=sources_dir)  # pyright: ignore[reportCallIssue]


def _stub_env(monkeypatch: pytest.MonkeyPatch, git_sources: str = "") -> None:
    """The resolver's env fallback, driven by a fresh ``Settings`` (the
    dev ``.env`` never leaks in — the task-02 pattern)."""
    monkeypatch.setattr(
        git_sources_resolver,
        "get_settings",
        lambda: Settings(_env_file=None, git_sources=git_sources),  # pyright: ignore[reportCallIssue]
    )


def _seed(db: Session, url: str) -> None:
    db.add(GitSource(url=url))
    db.commit()


def _seed_local(db: Session, path: Path) -> None:
    """A ``kind=local`` row as the phase-38 API stores it: the expanded
    absolute path in both ``path`` and the NOT-NULL ``url`` column."""
    db.add(GitSource(url=str(path), kind="local", path=str(path)))
    db.commit()


@pytest.fixture()
def clean_documents(db: Session) -> Iterator[None]:
    """The real-import tests write ``documents``/``chunks`` (the canonical
    KB state) — global, truncated around every such test."""
    db.execute(text("TRUNCATE chunks, documents"))
    db.commit()
    yield
    db.execute(text("TRUNCATE chunks, documents"))
    db.commit()


@pytest.fixture(autouse=True)
def _clean_folder_summaries(db: Session) -> Iterator[None]:
    """Phase 94: the ``folder_summaries`` table is global state the real
    generator (the real-import tests) writes — truncated around every
    sync test so the change-gate / table-empty-gate assertions start
    from a known (empty) table."""
    db.execute(text("TRUNCATE folder_summaries"))
    db.commit()
    yield
    db.execute(text("TRUNCATE folder_summaries"))
    db.commit()


def _real_llm(monkeypatch: pytest.MonkeyPatch) -> None:
    """The pipeline's ``LLMClient`` becomes the deterministic in-process
    ``FakeEmbedder`` (real import, no network). ``FakeEmbedder``
    implements ``embed_one`` + ``chat``, so the phase-41 probe passes
    against it without a stub."""
    monkeypatch.setattr(sync_api, "LLMClient", lambda: FakeEmbedder())


def _stub_probe(monkeypatch: pytest.MonkeyPatch) -> list[Any]:
    """Stub the phase-41 model probe (a real ``LLMClient`` in the runner
    would hit the network). Returns the clients the probe received, so
    tests can assert the probe and the import share one client."""
    seen: list[Any] = []

    async def fake_check_models(llm: LLMClient) -> None:
        seen.append(llm)

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


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 = 5.0) -> dict:
    """Poll ``GET /api/sync/status`` until ``state == want`` (terminal).

    Any state other than ``running`` before the deadline fails loudly —
    an unexpected ``failed`` must never be masked by the wait.
    """
    deadline = time.monotonic() + timeout
    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.05)
    raise AssertionError(f"sync did not reach {want!r} within {timeout}s")


def _wait_phase(client: TestClient, want: str, timeout: float = 5.0) -> dict:
    """Poll ``GET /api/sync/status`` until ``phase == want`` (the phase
    machine, phase 98 task 01), failing loudly if the run leaves
    ``running`` first.

    Phase 179 (task 02) is why this exists: the sync's git ops now run
    OFF the event loop (``asyncio.to_thread``), so how far the run has
    progressed by the time the client issues its NEXT request is
    genuinely concurrent — pre-fix the spawned task reached the import
    span before the trigger's own response went out, and a test that
    asserted a mid-run phase silently depended on that. Tests wait for
    the phase instead."""
    deadline = time.monotonic() + timeout
    while True:
        body = client.get("/api/sync/status").json()
        assert body["state"] == "running", (
            f"run left state {body['state']!r} before phase {want!r}: {body}"
        )
        if body["phase"] == want:
            return body
        if time.monotonic() >= deadline:
            raise AssertionError(
                f"sync did not report phase {want!r} within {timeout}s: {body}"
            )
        time.sleep(0.02)


class FakeImportSources:
    """Records every ``import_sources`` call; returns a canned summary."""

    def __init__(self, summary: ImportSummary, delay: float = 0.0) -> None:
        self.summary = summary
        self.delay = delay
        self.sources: list[list[Path]] = []
        self.llms: list[LLMClient] = []
        self.prune_flags: list[bool] = []
        # Phase 64 (task 02): the progress hook the runner passes (a live
        # closure while wired, None if the wiring regresses).
        self.progress_hooks: list[object] = []
        # Phase 89: the per-root ignore map the runner builds from the
        # rows' ``ignore_paths`` (keyed by the root string the importer
        # sees; two rows sharing a root string get the union).
        self.ignore_maps: list[dict[str, list[str]]] = []
        # Phase 105: the per-root hidden-folders flag map the runner
        # builds from the rows' ``include_hidden`` (same root-string
        # keying; a shared-root collision ORs — if either row says
        # "index hidden", the root does).
        self.include_hidden_maps: list[dict[str, bool]] = []
        # Phase 106: the per-root source-date map the runner builds
        # from ``file_commit_dates`` after each git clone (git rows
        # only, same root-string keying; local rows contribute
        # nothing).
        self.doc_dates_maps: list[dict[str, dict[str, datetime]]] = []
        # Phase 129 (task 03): the chunk-position hook the runner
        # passes (a live closure while wired, None if the wiring
        # regresses) — the canned fake never fires it (the runner's
        # status chunk fields stay 0/0, like a no-batch import).
        self.chunk_hooks: list[object] = []
        # Phase 152 (task 03): the image-position hook the runner
        # passes (a live closure while wired, None if the wiring
        # regresses) — the canned fake never fires it (the status
        # image fields stay 0/0, like the chunk fields of a no-batch
        # import).
        self.image_hooks: list[object] = []

    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,  # phase 129
        on_image_progress: Callable[[int, int], None] | None = None,  # phase 152
        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,  # phase 106
    ) -> ImportSummary:
        self.sources.append(list(sources))
        self.llms.append(llm)
        self.prune_flags.append(prune)
        self.progress_hooks.append(progress)
        self.chunk_hooks.append(on_chunk_progress)
        self.image_hooks.append(on_image_progress)
        self.ignore_maps.append(ignore_by_root or {})
        self.include_hidden_maps.append(include_hidden_by_root or {})
        self.doc_dates_maps.append(doc_dates_by_root or {})
        if self.delay:
            await asyncio.sleep(self.delay)
        return self.summary


class FakeOverview:
    """Records every ``regenerate_overview`` call; canned result."""

    def __init__(self, ok: bool = True) -> None:
        self.ok = ok
        self.llms: list[LLMClient] = []

    async def __call__(self, llm: LLMClient, session: Session | None = None) -> bool:
        self.llms.append(llm)
        return self.ok


class FakeFolderSummaries:
    """Records every ``generate_folder_summaries`` call; canned stats.

    Phase 94 (task 02): keeps the fake-import tests at the deterministic
    layer boundary — the real generator would read the global
    ``documents`` table and call the (real) ``LLMClient`` over the
    network. The generator only flushes, so the fake honours the
    ``skip`` flag the same way (the zero stats, no side effects).
    Phase 96 (task 03): the unchanged-walk gap path calls the
    generator with ``only_missing=True`` — the fake records the flag
    the same way it records ``skip`` (the real gap probe,
    ``missing_folder_summaries``, runs against the real tables).
    Phase 98 (task 01): records the ``on_progress`` hook the runner
    wires into the generation branches (a live closure while the
    wiring holds, None if it regresses); a canned *progress* list of
    ``(done, total, source, folder_path)`` steps is fired through the
    hook when given — the same way the import's file counter is driven
    through its hook, so the status's summary counters are testable
    deterministically.
    Phase 127 (task 03): records the ``changed_docs=`` change set the
    runner passes on the CHANGED-KB branch (the import's
    ``changed_paths`` — None when the runner regresses to the
    change-set-less full regeneration; the unchanged-walk gap-fill
    branch never passes it).
    """

    ZERO = {"generated": 0, "failed": 0, "pruned": 0}

    def __init__(
        self,
        stats: dict[str, int] | None = None,
        progress: list[tuple[int, int, str, str]] | None = None,
    ) -> None:
        self.stats = stats if stats is not None else dict(self.ZERO)
        self.progress_steps = list(progress or [])
        self.llms: list[LLMClient] = []
        self.sessions: list[Session] = []
        self.skip_flags: list[bool] = []
        self.only_missing_flags: list[bool] = []
        self.progress_hooks: list[Callable[[int, int, str, str], None] | None] = []
        self.changed_docs_sets: list[Sequence[tuple[str, str]] | None] = []

    async def __call__(
        self,
        db: Session,
        llm: LLMClient,
        *,
        skip: bool = False,
        only_missing: bool = False,
        changed_docs: Sequence[tuple[str, str]] | None = None,
        on_progress: Callable[[int, int, str, str], None] | None = None,
    ) -> dict[str, int]:
        self.skip_flags.append(skip)
        self.only_missing_flags.append(only_missing)
        self.changed_docs_sets.append(changed_docs)
        self.progress_hooks.append(on_progress)
        if skip:
            return dict(self.ZERO)
        self.llms.append(llm)
        self.sessions.append(db)
        for done, total, source, folder_path in self.progress_steps:
            if on_progress is not None:
                on_progress(done, total, source, folder_path)
        return dict(self.stats)


def _fake_clone() -> tuple[list[tuple[str, Path]], object]:
    """A ``clone_or_pull`` that materialises a checkout with one .md
    file. Phase 135: accepts the row's branch pin (the third argument
    — ``None`` for every row this file seeds)."""
    calls: list[tuple[str, Path]] = []

    def fake_clone_or_pull(url: str, dest: Path | str, branch: str | None = None) -> Path:
        dest = Path(dest)
        dest.mkdir(parents=True, exist_ok=True)
        (dest / "notes.md").write_text(f"# {dest.name}\ncontent for the KB\n", encoding="utf-8")
        calls.append((url, dest))
        return dest

    return calls, fake_clone_or_pull


# --- anonymous -------------------------------------------------------------


def test_anonymous_gets_403_on_both_endpoints(client: TestClient) -> None:
    r = client.get("/api/sync/status")
    assert r.status_code == 403
    assert r.json() == {"detail": "admin only"}
    r = client.post("/api/sync")
    assert r.status_code == 403
    assert r.json() == {"detail": "admin only"}


def test_anonymous_gets_403_on_cancel(client: TestClient) -> None:
    """Phase 134 (task 01): the cancel endpoint rides the router's
    ``require_admin`` dependency — anonymous gets the same 403 as the
    other sync surface (the completion criterion's 403 pin)."""
    r = client.post("/api/sync/cancel")
    assert r.status_code == 403
    assert r.json() == {"detail": "admin only"}


# --- admin: success --------------------------------------------------------


def test_admin_sync_success_reports_full_detail(
    sync_client: TestClient, monkeypatch: pytest.MonkeyPatch, db: Session, tmp_path: Path
) -> None:
    repo_url = f"file://{tmp_path / 'repo.git'}"
    _seed(db, repo_url)  # the phase-35 resolver picks the DB row up
    _stub_env(monkeypatch)  # env must not matter once the table has a row
    monkeypatch.setattr(
        sync_api,
        "get_settings",
        lambda: _settings(sources_dir=str(tmp_path / "bor")),
    )
    clone_calls, fake_clone = _fake_clone()
    monkeypatch.setattr(sync_api, "clone_or_pull", fake_clone)
    probe_seen = _stub_probe(monkeypatch)  # real LLMClient — probe stubbed
    summary = ImportSummary(
        files=5, added=1, updated=2, unchanged=2, pruned=3, errors=0,
        chunks=11, embed_batches=4, summaries=1, summary_errors=0,
    )
    fake_import = FakeImportSources(summary)
    monkeypatch.setattr(sync_api, "import_sources", fake_import)
    fake_overview = FakeOverview(ok=True)
    monkeypatch.setattr(sync_api, "regenerate_overview", fake_overview)
    # Phase 98 (task 01): the canned summary steps fire through the
    # runner's live hook closure — the terminal's summary counters
    # (kept per the keep-final-counts convention) are pinned below.
    fake_folders = FakeFolderSummaries(
        progress=[(1, 2, "repo", ""), (2, 2, "repo", "a")]
    )
    monkeypatch.setattr(sync_api, "generate_folder_summaries", fake_folders)

    _login(sync_client)
    assert sync_client.get("/api/sync/status").json() == {
        "state": "idle",
        "started_at": None,
        "finished_at": None,
        "detail": {},
        "error": None,
        # Phase 64 (task 02): the per-file progress keys — null/0/0 idle.
        "current_file": None,
        "files_done": 0,
        "files_total": 0,
        # Phase 129 (task 03): the chunk-position keys — 0/0 idle
        # (the dataclass defaults).
        "current_chunk": 0,
        "chunks_total": 0,
        # Phase 152 (task 01): the image-position keys — 0/0 idle
        # (the dataclass defaults).
        "images_done": 0,
        "images_total": 0,
        # Phase 98 (task 01): the phase-machine keys — null/0/0/0 idle.
        "phase": None,
        "current_summary": None,
        "summaries_done": 0,
        "summaries_total": 0,
    }

    r = sync_client.post("/api/sync")
    assert r.status_code == 202
    assert r.json() == {"detail": "sync started"}

    body = _poll(sync_client, "success")
    assert body["error"] is None
    # ISO-8601 timestamps round-trip; finished after started.
    started = datetime.fromisoformat(body["started_at"])
    finished = datetime.fromisoformat(body["finished_at"])
    assert finished >= started
    # Every ImportSummary field + the overview flag, verbatim.
    assert body["detail"] == {
        "files": 5, "added": 1, "updated": 2, "unchanged": 2, "pruned": 3,
        "errors": 0, "chunks": 11, "summaries": 1, "summary_errors": 0,
        "dates_updated": 0,  # phase 106: additive key, after summary_errors
        "overview": True,
        "sources_version": 1,  # phase 53: changed KB → exactly one bump (0 → 1)
    }
    # Phase 106: the runner feeds the per-root date map — the fake
    # checkout is not a git repo, so ``file_commit_dates`` fails soft
    # to ``{}`` for the one git row (root-keyed).
    assert fake_import.doc_dates_maps == [{str(tmp_path / "bor" / "repo"): {}}]
    # The bump committed: the counter advanced exactly once, not twice.
    assert current_sources_version(db) == 1
    # Git: the configured repo was cloned into BOR_SOURCES_DIR/<repo-name>/.
    assert clone_calls == [(repo_url, tmp_path / "bor" / "repo")]
    # Import: exactly the checkouts, with prune=True (the button is the
    # canonical "mirror the repos" action) and a real LLMClient.
    assert fake_import.sources == [[tmp_path / "bor" / "repo"]]
    assert fake_import.prune_flags == [True]
    assert len(fake_import.llms) == 1
    assert isinstance(fake_import.llms[0], LLMClient)
    # Phase 64 (task 02): the runner wires the per-file progress hook
    # (the live closure the status endpoint reads while the import runs).
    assert len(fake_import.progress_hooks) == 1
    assert callable(fake_import.progress_hooks[0])
    # Overview: refreshed (added + updated > 0) with the same client.
    assert fake_overview.llms == [fake_import.llms[0]]
    # Phase 41: the probe ran first and got the very client the import
    # and the overview reuse.
    assert probe_seen == [fake_import.llms[0]]
    # Phase 98 (task 01): the summary hook was wired into the
    # changed-KB branch and fired the canned steps — the terminal
    # clears phase + current_summary and KEEPS the hook's final
    # summary counts (the phase-64 keep-final-counts convention).
    assert len(fake_folders.progress_hooks) == 1
    assert fake_folders.progress_hooks[0] is not None
    assert body["phase"] is None
    assert body["current_summary"] is None
    assert body["summaries_done"] == 2 and body["summaries_total"] == 2


def test_unchanged_kb_skips_overview_refresh(
    sync_client: TestClient, monkeypatch: pytest.MonkeyPatch, db: Session, tmp_path: Path
) -> None:
    """Phase-31 trigger is change-gated: added + updated == 0 → no ``lite`` call."""
    repo_url = f"file://{tmp_path / 'repo.git'}"
    _seed(db, repo_url)
    _stub_env(monkeypatch)
    monkeypatch.setattr(
        sync_api,
        "get_settings",
        lambda: _settings(sources_dir=str(tmp_path / "bor")),
    )
    _, fake_clone = _fake_clone()
    monkeypatch.setattr(sync_api, "clone_or_pull", fake_clone)
    _stub_probe(monkeypatch)
    summary = ImportSummary(files=7, added=0, updated=0, unchanged=7, pruned=0)
    fake_import = FakeImportSources(summary)
    monkeypatch.setattr(sync_api, "import_sources", fake_import)
    fake_overview = FakeOverview(ok=True)
    monkeypatch.setattr(sync_api, "regenerate_overview", fake_overview)
    monkeypatch.setattr(sync_api, "generate_folder_summaries", FakeFolderSummaries())

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

    assert body["detail"]["overview"] is False
    assert fake_overview.llms == []  # no wasted model call
    assert len(fake_import.llms) == 1  # the import itself ran
    # Phase 53: an unchanged re-sync never bumps — the detail reports
    # the current (unadvanced) generation.
    assert body["detail"]["sources_version"] == 0
    assert current_sources_version(db) == 0


# --- admin: concurrency ----------------------------------------------------


def test_double_trigger_while_running_returns_409(
    sync_client: TestClient, monkeypatch: pytest.MonkeyPatch, db: Session, tmp_path: Path
) -> None:
    repo_url = f"file://{tmp_path / 'repo.git'}"
    _seed(db, repo_url)
    _stub_env(monkeypatch)
    monkeypatch.setattr(
        sync_api,
        "get_settings",
        lambda: _settings(sources_dir=str(tmp_path / "bor")),
    )
    _, fake_clone = _fake_clone()
    monkeypatch.setattr(sync_api, "clone_or_pull", fake_clone)
    _stub_probe(monkeypatch)
    # The in-flight run takes a while (asyncio.sleep) so the second POST
    # lands while it is still running (phase 179: 1.5 s, not 0.5 s — the
    # run now yields the loop at its git ops, so the window the second
    # POST and the phase wait fall into is wider on purpose).
    fake_import = FakeImportSources(ImportSummary(files=1, added=1), delay=1.5)
    monkeypatch.setattr(sync_api, "import_sources", fake_import)
    monkeypatch.setattr(sync_api, "regenerate_overview", FakeOverview(ok=True))
    monkeypatch.setattr(sync_api, "generate_folder_summaries", FakeFolderSummaries())

    _login(sync_client)
    assert sync_client.post("/api/sync").status_code == 202

    r = sync_client.post("/api/sync")  # second trigger while running
    assert r.status_code == 409
    assert r.json() == {"detail": "a sync is already running"}

    # Phase 179 (task 02): WAIT for the import phase — the run's
    # progress is concurrent now that the git ops left the loop, so the
    # next request no longer inherits a run that already ran past them.
    body = _wait_phase(sync_client, "import")
    assert body["state"] == "running"
    assert body["started_at"] is not None
    assert body["finished_at"] is None
    assert body["error"] is None
    # Phase 98 (task 01): the in-flight run (parked in the fake
    # import's delay) reports the import phase — the summary span has
    # not started, so its counters are 0/0.
    assert body["current_summary"] is None
    assert body["summaries_done"] == 0 and body["summaries_total"] == 0

    # The (single) run completes; the import ran exactly once.
    body = _poll(sync_client, "success")
    assert len(fake_import.sources) == 1
    # Phase 98 (task 01): the terminal cleared the phase keys (the
    # fake folder step fired no hook — nothing to keep).
    assert body["phase"] is None and body["current_summary"] is None
    assert body["summaries_done"] == 0 and body["summaries_total"] == 0


# --- phase 98 (task 01): the status's phase machine --------------------
# The background task runs on the app's event loop, so every gate below
# parks AWAY from the loop (``asyncio.to_thread(event.wait)``) — a
# blocking wait on the loop thread would deadlock the very status
# endpoint the test is polling.


class _GatedImport:
    """An import that fires the runner's progress hook once, then
    parks on a threading gate — the test reads the status mid-import
    (the phase ``"import"`` + the file hook).

    Phase 152 (task 03): optionally ALSO fires the in-flight file's
    image-position pairs (``image_pairs``) through the runner's OWN
    ``on_image_progress`` closure: pair one lands before the main park
    (readable mid-run), every later pair after the main park + its own
    park (``image_release``), so each position is readable on the
    status between polls (the phase-129 chunk idiom, mirrored for the
    vision-image position)."""

    def __init__(
        self,
        summary: ImportSummary,
        started: threading.Event,
        release: threading.Event,
        image_pairs: Sequence[tuple[int, int]] = (),
        image_release: threading.Event | None = None,
    ) -> None:
        self.summary = summary
        self.started = started
        self.release = release
        self.hook_calls: list[tuple[str, str, int, int]] = []
        # Phase 129 (task 03): the chunk-position hook the runner
        # passes (recorded for the wiring pin — the gated fake never
        # fires it, so the status chunk fields stay 0/0).
        self.chunk_hook: Callable[[int, int], None] | None = None
        # Phase 152 (task 03): the image-position hook the runner
        # passes (recorded for the wiring pin) + the optional image
        # pairs (see the class docstring).
        self.image_hook: Callable[[int, int], None] | None = None
        self.image_pairs = list(image_pairs)
        self.image_release = image_release
        self.image_hook_calls: list[tuple[int, int]] = []

    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,  # phase 129
        on_image_progress: Callable[[int, int], None] | None = None,  # phase 152
        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,  # phase 106
    ) -> ImportSummary:
        if progress is not None:
            progress("repo", "notes/deep.md", 1, 3)
            self.hook_calls.append(("repo", "notes/deep.md", 1, 3))
        self.chunk_hook = on_chunk_progress
        self.image_hook = on_image_progress
        if on_image_progress is not None and self.image_pairs:
            # Phase 152 (task 03): the in-flight PDF's image-position
            # pairs, fired through the runner's OWN hook closure (the
            # closure under test) — pair one before the main park, each
            # later pair after the main park + its own park (each park
            # lets the test read the status between polls). A wiring
            # regression to None skips this branch and the pair test
            # times out.
            on_image_progress(*self.image_pairs[0])
            self.image_hook_calls.append(self.image_pairs[0])
            self.started.set()
            await asyncio.to_thread(self.release.wait)
            for pair in self.image_pairs[1:]:
                on_image_progress(*pair)
                self.image_hook_calls.append(pair)
                if self.image_release is not None:
                    await asyncio.to_thread(self.image_release.wait)
            return self.summary
        self.started.set()
        await asyncio.to_thread(self.release.wait)
        return self.summary


class _GatedOverview:
    """A KB-overview step that parks on a threading gate once it
    starts — the test reads the status mid-overview (phase
    ``"overview"``)."""

    def __init__(self, started: threading.Event, release: threading.Event) -> None:
        self.started = started
        self.release = release

    async def __call__(self, llm: LLMClient, session: Session | None = None) -> bool:
        self.started.set()
        await asyncio.to_thread(self.release.wait)
        return True


class _GatedFolderSummaries:
    """A folder-summary step that fires the runner's progress hook
    (canned steps), parks mid-span — the (long) phase the user
    reported — then fires the final steps and returns canned stats.
    The test reads the status mid-span (the phase ``"summaries"`` +
    the hook's folder + counters)."""

    def __init__(
        self,
        started: threading.Event,
        release: threading.Event,
        steps_before: list[tuple[int, int, str, str]],
        steps_after: list[tuple[int, int, str, str]],
    ) -> None:
        self.started = started
        self.release = release
        self.steps_before = steps_before
        self.steps_after = steps_after
        self.hook: Callable[[int, int, str, str], None] | None = None
        self.changed_docs: Sequence[tuple[str, str]] | None = None

    async def __call__(
        self,
        db: Session,
        llm: LLMClient,
        *,
        skip: bool = False,
        only_missing: bool = False,
        changed_docs: Sequence[tuple[str, str]] | None = None,
        on_progress: Callable[[int, int, str, str], None] | None = None,
    ) -> dict[str, int]:
        assert not skip, "these runs take a generation branch, never --limit"
        self.hook = on_progress
        self.changed_docs = changed_docs
        for done, total, source, folder_path in self.steps_before:
            if on_progress is not None:
                on_progress(done, total, source, folder_path)
        self.started.set()
        await asyncio.to_thread(self.release.wait)
        for done, total, source, folder_path in self.steps_after:
            if on_progress is not None:
                on_progress(done, total, source, folder_path)
        return {"generated": 3, "failed": 0, "pruned": 0, "kept_manual": 0}


def test_status_reports_the_phase_machine_across_the_run(
    sync_client: TestClient, monkeypatch: pytest.MonkeyPatch, db: Session, tmp_path: Path
) -> None:
    """Phase 98 (task 01), the full state machine: idle reports the
    four keys null/0/0/0; the model-check + clone/pull prelude
    reports ``phase: null`` (D1 — the bare label's pin); the import
    reports ``"import"`` with the file hook (the phase-64 shape,
    unchanged); the KB-overview step reports ``"overview"``; the
    folder-summary span reports ``"summaries"`` with the hook's
    folder + done/total (the import's final file position kept — the
    pause the user reported); the success terminal clears ``phase`` +
    ``current_summary`` and KEEPS the hook's final summary counts
    (the phase-64 keep-final-counts convention)."""
    repo_url = f"file://{tmp_path / 'repo.git'}"
    _seed(db, repo_url)
    _stub_env(monkeypatch)
    monkeypatch.setattr(
        sync_api,
        "get_settings",
        lambda: _settings(sources_dir=str(tmp_path / "bor")),
    )
    _, fake_clone = _fake_clone()
    monkeypatch.setattr(sync_api, "clone_or_pull", fake_clone)
    _stub_probe(monkeypatch)
    gated_import = _GatedImport(
        ImportSummary(files=3, added=1, updated=1, unchanged=1),
        threading.Event(),
        threading.Event(),
    )
    monkeypatch.setattr(sync_api, "import_sources", gated_import)
    gated_overview = _GatedOverview(threading.Event(), threading.Event())
    monkeypatch.setattr(sync_api, "regenerate_overview", gated_overview)
    gated_folders = _GatedFolderSummaries(
        threading.Event(),
        threading.Event(),
        steps_before=[(1, 3, "repo", "")],
        steps_after=[(3, 3, "repo", "notes")],
    )
    monkeypatch.setattr(sync_api, "generate_folder_summaries", gated_folders)

    _login(sync_client)
    # Idle: the four phase keys ride along as null/0/0/0, and the
    # phase-129 chunk keys as 0/0 (the dataclass defaults).
    body = sync_client.get("/api/sync/status").json()
    assert body["state"] == "idle"
    assert body["phase"] is None
    assert body["current_summary"] is None
    assert body["summaries_done"] == 0 and body["summaries_total"] == 0
    assert body["current_chunk"] == 0 and body["chunks_total"] == 0

    assert sync_client.post("/api/sync").status_code == 202
    try:
        assert gated_import.started.wait(5.0), "the run never reached the import"
        s = sync_client.get("/api/sync/status").json()
        assert s["state"] == "running"
        assert s["phase"] == "import"
        assert s["current_file"] == "repo/notes/deep.md"  # the file hook as today
        assert s["files_done"] == 1 and s["files_total"] == 3
        assert s["current_summary"] is None
        assert s["summaries_done"] == 0 and s["summaries_total"] == 0
        # Phase 129 (task 03): no embed batch has fired the chunk hook
        # (the gated fake never does) — the position stays 0/0.
        assert s["current_chunk"] == 0 and s["chunks_total"] == 0
        gated_import.release.set()
        assert gated_overview.started.wait(5.0), "the run never reached the overview"
        s = sync_client.get("/api/sync/status").json()
        assert s["state"] == "running"
        assert s["phase"] == "overview"
        assert s["current_summary"] is None
        # Phase 129 (task 03): the import→overview transition cleared
        # the chunk position (the convention split — the FILE counts
        # below keep their final values).
        assert s["current_chunk"] == 0 and s["chunks_total"] == 0
        gated_overview.release.set()
        assert gated_folders.started.wait(5.0), "the run never reached the summaries"
        s = sync_client.get("/api/sync/status").json()
        assert s["state"] == "running"
        assert s["phase"] == "summaries"
        # The hook's first step: the source-root row (the bare source
        # name — folder_path "" never gets a slash).
        assert s["current_summary"] == "repo"
        assert s["summaries_done"] == 1 and s["summaries_total"] == 3
        # The import's final position is kept through the summary span
        # (the "number pauses" span — the user's report).
        assert s["files_done"] == 1 and s["files_total"] == 3
        # Phase 129 (task 03): the import→summaries transition kept
        # the chunk position cleared (0/0 through the summary span).
        assert s["current_chunk"] == 0 and s["chunks_total"] == 0
        gated_folders.release.set()
    finally:
        gated_import.release.set()
        gated_overview.release.set()
        gated_folders.release.set()

    body = _poll(sync_client, "success")
    # Terminal: phase + current_summary cleared ... (the final step
    # fired post-park, so the kept counts are its position).
    assert body["phase"] is None
    assert body["current_summary"] is None
    assert body["summaries_done"] == 3 and body["summaries_total"] == 3
    # Phase 129 (task 03): the terminal keeps the chunk position
    # cleared (0/0) — the wiring pin: the runner passed a live chunk
    # closure to the import (it just never fired in this canned run).
    assert body["current_chunk"] == 0 and body["chunks_total"] == 0
    assert gated_import.chunk_hook is not None
    # Wiring: the summary hook was the runner's live closure (the
    # counters above came through it); the file hook fired once.
    assert gated_folders.hook is not None
    assert gated_import.hook_calls == [("repo", "notes/deep.md", 1, 3)]


def test_image_position_climbs_between_polls_mid_run(
    sync_client: TestClient, monkeypatch: pytest.MonkeyPatch, db: Session, tmp_path: Path
) -> None:
    """Phase 152 (task 03): while a PDF's vision span is in flight,
    the status carries the in-flight image position — the pairs are
    fired through the runner's OWN ``_image_hook`` closure (task 03's
    wiring: ``import_sources(..., on_image_progress=_image_hook)``),
    and the position CLIMBS between two polls of
    ``GET /api/sync/status`` (1/15 → 7/15 — monotonic, never above
    the total, no backwards jumps — the owner's stuck label now
    moves), paired with the file the ``progress`` hook named (the
    hooks are paired — the position belongs to that file); the
    success terminal clears the position to 0/0 while the FILE
    counts keep their final values (the convention split — same as
    the chunk position)."""
    repo_url = f"file://{tmp_path / 'repo.git'}"
    _seed(db, repo_url)
    _stub_env(monkeypatch)
    monkeypatch.setattr(
        sync_api,
        "get_settings",
        lambda: _settings(sources_dir=str(tmp_path / "bor")),
    )
    _, fake_clone = _fake_clone()
    monkeypatch.setattr(sync_api, "clone_or_pull", fake_clone)
    _stub_probe(monkeypatch)
    image_release = threading.Event()
    gated_import = _GatedImport(
        ImportSummary(files=3, added=1, updated=1, unchanged=1),
        threading.Event(),
        threading.Event(),
        image_pairs=[(1, 15), (7, 15)],  # two describe calls of a 15-image span
        image_release=image_release,
    )
    monkeypatch.setattr(sync_api, "import_sources", gated_import)
    monkeypatch.setattr(sync_api, "regenerate_overview", FakeOverview(ok=True))
    monkeypatch.setattr(sync_api, "generate_folder_summaries", FakeFolderSummaries())

    _login(sync_client)
    # Idle: the image keys ride along as 0/0 (the dataclass defaults).
    body = sync_client.get("/api/sync/status").json()
    assert body["state"] == "idle"
    assert body["images_done"] == 0 and body["images_total"] == 0

    assert sync_client.post("/api/sync").status_code == 202
    try:
        assert gated_import.started.wait(5.0), "the run never reached the first image pair"
        # Poll 1 — the import phase, pair one live: the position is
        # 1/15, paired with the file the progress hook named; the
        # chunk position is 0/0 (this canned file has no embed
        # batches — the two positions are independent).
        s = sync_client.get("/api/sync/status").json()
        assert s["state"] == "running"
        assert s["phase"] == "import"
        assert s["current_file"] == "repo/notes/deep.md"
        assert s["files_done"] == 1 and s["files_total"] == 3
        assert s["current_chunk"] == 0 and s["chunks_total"] == 0
        assert s["images_done"] == 1 and s["images_total"] == 15
        # Release pair one: the import fires pair two (7/15 — the
        # NEXT describe call of the same 15-image span) and parks
        # again.
        gated_import.release.set()
        deadline = time.monotonic() + 5.0
        while time.monotonic() < deadline:
            s = sync_client.get("/api/sync/status").json()
            if s["images_done"] == 7:
                break
            assert s["state"] == "running", (
                f"unexpected state {s['state']!r} while waiting for pair two: {s}"
            )
            time.sleep(0.05)
        else:
            pytest.fail("the second image pair never landed on the status")
        # Poll 2 — the position CLIMBED (1 → 7, monotonic — no
        # backwards jump, never above the total): the label is
        # moving, not frozen.
        assert s["state"] == "running"
        assert s["phase"] == "import"
        assert s["current_file"] == "repo/notes/deep.md"
        assert s["images_done"] == 7 and s["images_total"] == 15
        assert 0 < s["images_done"] <= s["images_total"]
        image_release.set()
    finally:
        gated_import.release.set()
        image_release.set()

    body = _poll(sync_client, "success")
    # Terminal: the image position clears (0/0) ... the FILE counts
    # KEEP their final values (the convention split — the position is
    # a sub-position of a done file, like the chunk position).
    assert body["images_done"] == 0 and body["images_total"] == 0
    assert body["current_file"] is None
    assert body["files_done"] == 1 and body["files_total"] == 3
    # Wiring: BOTH pairs went through the runner's OWN live closure
    # (the status reads above came through it — a wiring regression
    # to None would have skipped the firing branch and timed out),
    # and the file hook fired once.
    assert gated_import.image_hook is not None
    assert gated_import.image_hook_calls == [(1, 15), (7, 15)]
    assert gated_import.hook_calls == [("repo", "notes/deep.md", 1, 3)]


# --- phase 152 (task 04): the full-run image-position monotonicity -------


def _kb_solid_png(value: int) -> bytes:
    """A 40x30 solid-color PNG (channel *value*): every value used
    below stays under the phase-151 blank gate's 235 mean floor
    (kept, never blank-skipped), and distinct values are distinct
    images (the parser's dedupe keeps every one)."""
    pix = fitz.Pixmap(fitz.csRGB, fitz.IRect(0, 0, 40, 30), 0)
    pix.clear_with(value)
    return pix.tobytes("png")


def _kb_image_pdf(path: Path, values: Sequence[int]) -> None:
    """A REAL one-page PDF (the fixture KB's image-bearing file): a
    text block comfortably over the phase-140 100-char sparseness
    bar (a USABLE layer — the phase-133 gate stays closed, zero
    pictorial candidates even with the toggle on — the "no other
    vision span" class) carrying *values* the solid-color embedded
    images (one placement each, distinct)."""
    doc = fitz.open()
    page = doc.new_page()
    text = (
        "The quadlet unit file defines the container interface, the "
        "image, the network, and the restart policy. Every key maps "
        "to a docker flag in the generated service file, and comments "
        "use the usual semicolon form throughout the whole document."
    )
    for i, line in enumerate(textwrap.wrap(text, 60)):
        page.insert_text((72, 72 + i * 14), line)
    y = 300
    for value in values:
        page.insert_image(fitz.Rect(72, y, 112, y + 30), stream=_kb_solid_png(value))
        y += 40
    doc.save(path)
    doc.close()


class _GatedVisionLLM(FakeEmbedder):
    """The phase-152 (task 04) mock LLM — the deterministic
    in-process ``FakeEmbedder`` whose VISION calls (the multimodal
    describe shape ``describe_image`` sends) PARK on a threading
    gate: the run parks inside the PDF's embedded-image span, so the
    test can read the status between the describe calls
    (deterministic — no sleeps, no wall-clock dependence; the gate
    holds the run until the test releases it). The hook fired BEFORE
    the parked call (the task-02 contract), so the position is live
    while the run is parked. Text (``lite``) calls keep the
    ``FakeEmbedder`` behaviour; the embeds are the deterministic
    vectors."""

    def __init__(self) -> None:
        super().__init__()
        self.vision_parks = 0  # the vision calls that parked (and released)
        self.vision_release = threading.Event()

    async def chat(
        self, messages: list[dict[str, Any]], model: str | None = None
    ) -> str:
        user = next((m["content"] for m in messages if m.get("role") == "user"), "")
        if isinstance(user, list):
            # One slow vision describe call — the gate (clear before
            # the wait — a latching set would otherwise leak across
            # calls): the test releases after reading the position.
            self.vision_release.clear()
            self.vision_parks += 1
            await asyncio.to_thread(self.vision_release.wait)
            return "A solid color swatch described."  # deterministic, non-empty
        return await super().chat(messages, model)


def test_image_position_monotonic_across_polls_over_a_real_pdf(
    sync_client: TestClient,
    monkeypatch: pytest.MonkeyPatch,
    db: Session,
    tmp_path: Path,
    clean_documents: None,
) -> None:
    """Phase 152 (task 04): a FULL sync run over a fixture KB with
    ONE image-bearing PDF (the REAL ``import_sources`` — the
    runner's ``_image_hook`` is driven by the importer's own
    embedded-image span; the LLM is the deterministic mock whose
    vision calls park on a gate) — polling ``GET /api/sync/status``
    from the test while the run is parked between the describe
    calls, the observed ``images_done`` sequence is MONOTONICALLY
    NON-DECREASING, never above ``images_total``, and ends at
    ``images_total`` (no backwards jumps, no overshoot — the
    owner's stuck label climbs one describe call at a time); the
    position is paired with the file the ``progress`` hook named
    (the hooks are paired), the success terminal clears it to 0/0
    while the FILE counts keep their final values, and the KB
    carries the PDF + its N image docs (the span really ran)."""
    kb = tmp_path / "PictKB"
    kb.mkdir()
    (kb / "notes.md").write_text("# Notes\nthe local fixture\n", encoding="utf-8")
    _kb_image_pdf(kb / "report.pdf", (90, 130, 170))  # 3 distinct images
    n = 3
    _seed_local(db, kb)
    _stub_env(monkeypatch)
    monkeypatch.setattr(
        sync_api,
        "get_settings",
        lambda: _settings(sources_dir=str(tmp_path / "bor")),
    )
    clone_calls, fake_clone = _fake_clone()
    monkeypatch.setattr(sync_api, "clone_or_pull", fake_clone)
    llm = _GatedVisionLLM()
    llm.settings = Settings(
        _env_file=None,  # pyright: ignore[reportCallIssue]
        import_extensions="md,txt,pdf",
        images=True,
        image_dir=str(tmp_path / "images"),
        pdf_page_dir=str(tmp_path / "pages"),
    )
    monkeypatch.setattr(sync_api, "LLMClient", lambda: llm)
    _stub_probe(monkeypatch)  # the probe passes on the mock (no network)

    _login(sync_client)

    def status() -> dict:
        return sync_client.get("/api/sync/status").json()

    def wait_for(predicate, what: str, timeout: float = 10.0) -> None:
        deadline = time.monotonic() + timeout
        while time.monotonic() < deadline:
            if predicate():
                return
            time.sleep(0.02)
        raise AssertionError(f"timed out waiting for {what}")

    # Idle: the image keys ride along as 0/0 (the dataclass
    # defaults).
    body = status()
    assert body["state"] == "idle"
    assert body["images_done"] == 0 and body["images_total"] == 0

    assert sync_client.post("/api/sync").status_code == 202
    try:
        # The run reaches the PDF's vision span: the first hook fired
        # (total = the span's N) and the first describe call parked
        # the run (the gate).
        wait_for(
            lambda: status()["state"] == "running"
            and status()["phase"] == "import"
            and status()["images_total"] == n,
            "the run to reach the PDF's vision span",
        )
        # Each describe call: the hook fired BEFORE the call, so the
        # position is live while the run is parked in the call —
        # observe it, release the call, let the next position climb.
        observed: list[tuple[int, int]] = []
        for expected in range(1, n + 1):
            def position_ready(target: int = expected) -> bool:
                # the default argument binds the loop variable (B023)
                return status()["images_done"] == target

            wait_for(position_ready, f"images_done to climb to {expected}")
            s = status()
            assert s["state"] == "running" and s["phase"] == "import"
            # The position is paired with the file the progress hook
            # named (the two hooks are paired — the position belongs
            # to that file).
            assert s["current_file"] == "PictKB/report.pdf"
            observed.append((s["images_done"], s["images_total"]))
            llm.vision_release.set()
        # The run finishes the rest of the pipeline (the span is
        # done — no more vision calls) and settles.
        assert llm.vision_parks == n, "one park per describe call"
    finally:
        llm.vision_release.set()

    body = _poll(sync_client, "success")
    # The observed sequence: monotonically non-decreasing, never
    # above the total, ending at the total (the gate made it exact —
    # 1..N, no backwards jumps, no overshoot).
    dones = [d for d, _ in observed]
    assert dones == list(range(1, n + 1)), f"the sequence: {observed}"
    assert all(d <= t for d, t in observed), "no overshoot"
    assert dones[-1] == n, "ends at images_total"
    assert {t for _, t in observed} == {n}, "one total for the whole span"
    # Terminal: the image position clears (0/0) ... the FILE counts
    # KEEP their final values (the convention split — the position
    # is a sub-position of a done file, like the chunk position).
    assert body["images_done"] == 0 and body["images_total"] == 0
    assert body["current_file"] is None
    assert body["files_done"] == 2 and body["files_total"] == 2
    assert clone_calls == []  # nothing to clone — local is walked directly
    # The KB carries the fixture md + the PDF + its N image docs —
    # the span really ran (N new image documents, no failures).
    assert body["detail"]["added"] == 2 + n
    assert body["detail"]["errors"] == 0
    docs = {(d["source"], d["path"])
           for d in sync_client.get("/api/docs").json()["documents"]}
    assert ("PictKB", "report.pdf") in docs
    assert sum(1 for _, p in docs if "#page=" in p) == n


def test_failed_terminal_after_the_summary_hook_clears_phase_keeps_counts(
    sync_client: TestClient, monkeypatch: pytest.MonkeyPatch, db: Session, tmp_path: Path
) -> None:
    """Phase 98 (task 01), the except terminal path: a run that dies
    AFTER the summary hook has fired (here: the folder step raises
    post-hook) clears ``phase`` + ``current_summary`` in the failed
    terminal AND keeps the hook's final summary counts next to the
    error — the same keep-final-counts convention as success (D1:
    BOTH terminal paths)."""
    repo_url = f"file://{tmp_path / 'repo.git'}"
    _seed(db, repo_url)
    _stub_env(monkeypatch)
    monkeypatch.setattr(
        sync_api,
        "get_settings",
        lambda: _settings(sources_dir=str(tmp_path / "bor")),
    )
    _, fake_clone = _fake_clone()
    monkeypatch.setattr(sync_api, "clone_or_pull", fake_clone)
    _stub_probe(monkeypatch)
    fake_import = FakeImportSources(
        ImportSummary(files=3, added=1, updated=1, unchanged=1)
    )
    monkeypatch.setattr(sync_api, "import_sources", fake_import)
    monkeypatch.setattr(sync_api, "regenerate_overview", FakeOverview(ok=True))

    class _HookThenFail:
        """Fires two progress steps through the runner's hook, then
        raises — the failure lands AFTER the hook moved the
        counters (the keep-counts path the success test pins for the
        happy terminal)."""

        def __init__(self) -> None:
            self.hook: Callable[[int, int, str, str], None] | None = None

        async def __call__(
            self,
            db: Session,
            llm: LLMClient,
            *,
            skip: bool = False,
            only_missing: bool = False,
            changed_docs: Sequence[tuple[str, str]] | None = None,  # phase 127
            on_progress: Callable[[int, int, str, str], None] | None = None,
        ) -> dict[str, int]:
            self.hook = on_progress
            assert on_progress is not None, "the runner must wire the hook"
            on_progress(1, 3, "repo", "")
            on_progress(2, 3, "repo", "notes")
            raise RuntimeError("simulated post-hook failure (test sentinel)")

    failing = _HookThenFail()
    monkeypatch.setattr(sync_api, "generate_folder_summaries", failing)

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

    assert "simulated post-hook failure" in body["error"]
    assert body["phase"] is None  # cleared in the failed terminal
    assert body["current_summary"] is None
    # The hook's FINAL summary counts survive the failure (D1).
    assert body["summaries_done"] == 2 and body["summaries_total"] == 3
    assert failing.hook is not None  # the wiring held


# --- phase 127 (task 03): the changed-KB branch's incremental wiring ----


def test_changed_sync_passes_the_change_set_to_the_generator(
    sync_client: TestClient,
    monkeypatch: pytest.MonkeyPatch,
    db: Session,
    tmp_path: Path,
    caplog: pytest.LogCaptureFixture,
) -> None:
    """Phase 127 (task 03): a changed-KB sync passes the import's
    change set into the generator (``changed_docs=summary.
    changed_paths`` — the incremental-regression restriction), NOT the
    gap-fill flag, the success ``detail`` keeps its exact pre-phase key
    set (the stats stay log-only), and the log line names the change
    count (``changed=N``)."""
    repo_url = f"file://{tmp_path / 'repo.git'}"
    _seed(db, repo_url)
    _stub_env(monkeypatch)
    monkeypatch.setattr(
        sync_api,
        "get_settings",
        lambda: _settings(sources_dir=str(tmp_path / "bor")),
    )
    _, fake_clone = _fake_clone()
    monkeypatch.setattr(sync_api, "clone_or_pull", fake_clone)
    _stub_probe(monkeypatch)
    # One updated document — the import's change set carries exactly
    # its (source, path) identity (task 01's contract).
    change_set = [("repo", "notes/deep.md")]
    summary = ImportSummary(
        files=4, added=0, updated=1, unchanged=3, changed_paths=change_set
    )
    fake_import = FakeImportSources(summary)
    monkeypatch.setattr(sync_api, "import_sources", fake_import)
    monkeypatch.setattr(sync_api, "regenerate_overview", FakeOverview(ok=True))
    fake_folders = FakeFolderSummaries()
    monkeypatch.setattr(sync_api, "generate_folder_summaries", fake_folders)

    _login(sync_client)
    with caplog.at_level(logging.INFO, logger="app.api.sync"):
        assert sync_client.post("/api/sync").status_code == 202
        body = _poll(sync_client, "success")

    # The CHANGED-KB branch fired the generator exactly once ... with
    # the change set, NOT the gap-fill flag (the two restriction
    # kwargs are orthogonal — no caller combines them).
    assert len(fake_folders.changed_docs_sets) == 1
    assert fake_folders.changed_docs_sets == [change_set]
    assert fake_folders.only_missing_flags == [False]
    assert fake_folders.skip_flags == [False]
    # The success detail is byte-identical in shape to pre-phase 127 —
    # no new key (the stats stay log-only, the phase-98 convention).
    assert set(body["detail"]) == {
        "files", "added", "updated", "unchanged", "pruned", "errors",
        "chunks", "summaries", "summary_errors", "dates_updated",
        "overview", "sources_version",
    }
    assert body["detail"]["added"] == 0 and body["detail"]["updated"] == 1
    assert body["detail"]["overview"] is True
    assert body["detail"]["sources_version"] == 1  # changed KB → one bump
    # The ample-logging line names the change count (PLAN §9).
    assert any(
        r.getMessage() == (
            "sync: folder_summaries changed=1 "
            "stats={'generated': 0, 'failed': 0, 'pruned': 0}"
        )
        for r in caplog.records
    )


def test_unchanged_walk_gap_probe_passes_no_change_set(
    sync_client: TestClient,
    monkeypatch: pytest.MonkeyPatch,
    db: Session,
    tmp_path: Path,
    clean_documents: None,
) -> None:
    """Phase 127 (task 03): the unchanged-walk gap probe is
    ORTHOGONAL to the change set — a walk that changed nothing with a
    real gap (seeded catalogue, empty ``folder_summaries`` table) still
    calls the generator exactly as before: ``only_missing=True`` and
    NO ``changed_docs`` (the gap probe's total is the missing count,
    as before)."""
    repo_url = f"file://{tmp_path / 'repo.git'}"
    _seed(db, repo_url)
    _stub_env(monkeypatch)
    monkeypatch.setattr(
        sync_api,
        "get_settings",
        lambda: _settings(sources_dir=str(tmp_path / "bor")),
    )
    _, fake_clone = _fake_clone()
    monkeypatch.setattr(sync_api, "clone_or_pull", fake_clone)
    _stub_probe(monkeypatch)
    # The gap: two docs under notes/ (candidates root + notes), the
    # folder_summaries table empty (the autouse fixture truncated it).
    db.add_all([
        Document(
            source="repo", path="notes/one.md", full_path="/tmp/one.md",
            title="One", content="# One\nfirst\n", content_hash="a" * 64,
        ),
        Document(
            source="repo", path="notes/two.md", full_path="/tmp/two.md",
            title="Two", content="# Two\nsecond\n", content_hash="b" * 64,
        ),
    ])
    db.commit()
    summary = ImportSummary(files=2, added=0, updated=0, unchanged=2)
    fake_import = FakeImportSources(summary)
    monkeypatch.setattr(sync_api, "import_sources", fake_import)
    monkeypatch.setattr(sync_api, "regenerate_overview", FakeOverview(ok=True))
    fake_folders = FakeFolderSummaries()
    monkeypatch.setattr(sync_api, "generate_folder_summaries", fake_folders)

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

    # The gap probe fired the generator exactly once — the gap-fill
    # shape, byte-identical to pre-phase 127: only_missing=True, and
    # NO change set (the gap is a stored-rows gap, not a document
    # change — passing changed_paths here would restrict the fill to
    # nothing).
    assert len(fake_folders.changed_docs_sets) == 1
    assert fake_folders.changed_docs_sets == [None]
    assert fake_folders.only_missing_flags == [True]
    assert body["detail"]["overview"] is False  # unchanged → no overview
    assert body["detail"]["sources_version"] == 0  # unchanged → no bump


class _AffectedKeysFolderSummaries:
    """A folder-summary step that mirrors the REAL generator's hook
    contract on a changed set (phase 127, task 02): the loop runs over
    the sorted AFFECTED keys (``affected_folder_keys(changed_docs)``),
    the hook fires once per key with ``total == len(keys)``, and the
    step parks mid-span after the first key — the (long) phase the
    user reported. The test reads the status mid-span and asserts the
    status's ``summaries_total`` IS the affected count (the hook's
    total), not the table's candidate count."""

    def __init__(
        self,
        started: threading.Event,
        release: threading.Event,
    ) -> None:
        self.started = started
        self.release = release
        self.changed_docs: Sequence[tuple[str, str]] | None = None
        self.keys: list[tuple[str, str]] = []
        self.hook: Callable[[int, int, str, str], None] | None = None

    async def __call__(
        self,
        db: Session,
        llm: LLMClient,
        *,
        skip: bool = False,
        only_missing: bool = False,
        changed_docs: Sequence[tuple[str, str]] | None = None,
        on_progress: Callable[[int, int, str, str], None] | None = None,
    ) -> dict[str, int]:
        assert not skip, "these runs take a generation branch, never --limit"
        self.hook = on_progress
        self.changed_docs = changed_docs
        assert on_progress is not None, "the runner must wire the hook"
        # The real generator's step 2: sorted(candidates & affected) —
        # the fake takes the affected side (the candidates are a
        # superset here: every affected folder holds a doc).
        self.keys = sorted(affected_folder_keys(changed_docs or []))
        assert self.keys, "the change set must affect at least the root"
        on_progress(1, len(self.keys), self.keys[0][0], self.keys[0][1])
        self.started.set()
        await asyncio.to_thread(self.release.wait)
        for i, (source, folder_path) in enumerate(self.keys[1:], start=2):
            on_progress(i, len(self.keys), source, folder_path)
        return {"generated": len(self.keys), "failed": 0, "pruned": 0}


def test_summaries_phase_reports_the_affected_count_as_total(
    sync_client: TestClient,
    monkeypatch: pytest.MonkeyPatch,
    db: Session,
    tmp_path: Path,
) -> None:
    """Phase 127 (task 03): mid-span, the status's ``summaries_total``
    is the AFFECTED count — ``len(affected_folder_keys(changed set))``
    (the hook's total), not the table's candidate count — and the
    terminal keeps that final count (the phase-64 convention, now over
    the affected span). The change set: one nested-folder doc (a
    three-key chain) + one root-level doc (shares the root key) →
    exactly 3 affected keys."""
    repo_url = f"file://{tmp_path / 'repo.git'}"
    _seed(db, repo_url)
    _stub_env(monkeypatch)
    monkeypatch.setattr(
        sync_api,
        "get_settings",
        lambda: _settings(sources_dir=str(tmp_path / "bor")),
    )
    _, fake_clone = _fake_clone()
    monkeypatch.setattr(sync_api, "clone_or_pull", fake_clone)
    _stub_probe(monkeypatch)
    change_set = [("repo", "notes/a/one.md"), ("repo", "top.md")]
    fake_import = FakeImportSources(
        ImportSummary(
            files=6, added=1, updated=2, unchanged=3, changed_paths=change_set
        )
    )
    monkeypatch.setattr(sync_api, "import_sources", fake_import)
    monkeypatch.setattr(sync_api, "regenerate_overview", FakeOverview(ok=True))
    gated_folders = _AffectedKeysFolderSummaries(
        threading.Event(), threading.Event()
    )
    monkeypatch.setattr(sync_api, "generate_folder_summaries", gated_folders)

    _login(sync_client)
    assert sync_client.post("/api/sync").status_code == 202
    try:
        assert gated_folders.started.wait(5.0), "the run never reached the summaries"
        s = sync_client.get("/api/sync/status").json()
        assert s["state"] == "running"
        assert s["phase"] == "summaries"
        # The hook's first key: the source root (the bare source name —
        # folder_path "" never gets a slash) ...
        assert s["current_summary"] == "repo"
        assert s["summaries_done"] == 1
        # ... and the total IS the affected count (the task's pin):
        # notes/a/one.md → ("repo", ""), ("repo", "notes"),
        # ("repo", "notes/a"); top.md → ("repo", "") — 3 keys.
        expected = sorted(affected_folder_keys(change_set))
        assert expected == [("repo", ""), ("repo", "notes"), ("repo", "notes/a")]
        assert s["summaries_total"] == len(expected)
        assert s["summaries_total"] == 3
        gated_folders.release.set()
    finally:
        gated_folders.release.set()

    body = _poll(sync_client, "success")
    # Terminal: phase + current_summary cleared, the hook's FINAL
    # (affected) counts kept (the phase-64 keep-final-counts rule, now
    # over the affected span).
    assert body["phase"] is None and body["current_summary"] is None
    assert body["summaries_done"] == 3 and body["summaries_total"] == 3
    # Wiring: the runner passed the change set, not the gap-fill flag.
    assert gated_folders.changed_docs == change_set
    assert gated_folders.hook is not None


# --- admin: failures -------------------------------------------------------


def test_git_failure_marks_failed_and_skips_import(
    sync_client: TestClient, monkeypatch: pytest.MonkeyPatch, db: Session, tmp_path: Path
) -> None:
    repo_url = f"file://{tmp_path / 'bad.git'}"
    _seed(db, repo_url)
    _stub_env(monkeypatch)
    monkeypatch.setattr(
        sync_api,
        "get_settings",
        lambda: _settings(sources_dir=str(tmp_path / "bor")),
    )

    def failing_clone(url: str, dest: Path | str, branch: str | None = None) -> Path:
        raise GitSyncError(
            f"git clone {url} failed (exit 128): "
            "fatal: repository not found"
        )

    monkeypatch.setattr(sync_api, "clone_or_pull", failing_clone)
    _stub_probe(monkeypatch)
    fake_import = FakeImportSources(ImportSummary())
    monkeypatch.setattr(sync_api, "import_sources", fake_import)
    fake_overview = FakeOverview(ok=True)
    monkeypatch.setattr(sync_api, "regenerate_overview", fake_overview)

    _login(sync_client)
    assert sync_client.post("/api/sync").status_code == 202

    body = _poll(sync_client, "failed")
    assert "bad.git" in body["error"]  # the failing repo is named
    assert "fatal: repository not found" in body["error"]
    assert body["detail"] == {}
    assert body["finished_at"] is not None
    # Phase 98 (task 01): the failed terminal (died in the clone/pull
    # prelude — ``phase`` was still null) clears the phase keys; the
    # run never reached the import or summary spans, so all counters
    # are 0/0.
    assert body["phase"] is None
    assert body["current_summary"] is None
    assert body["current_file"] is None
    assert body["files_done"] == 0 and body["files_total"] == 0
    assert body["summaries_done"] == 0 and body["summaries_total"] == 0
    assert fake_import.sources == []  # no partial import
    assert fake_overview.llms == []
    # Phase 53: a FAILED sync never bumps — the version is untouched.
    assert current_sources_version(db) == 0

    # A failed run leaves the system restartable: a new POST is accepted.
    assert sync_client.post("/api/sync").status_code == 202
    _poll(sync_client, "failed")


def test_no_sources_configured_fails_loudly(
    sync_client: TestClient, monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
    """Both origins empty — no git rows, no local rows, no env URLs
    (the truncate fixture + a blank env) → the fail-loud error
    (phase 38: git-only message retired)."""
    # Whitespace-only is just as unconfigured as empty.
    _stub_env(monkeypatch, " , ")
    monkeypatch.setattr(
        sync_api,
        "get_settings",
        lambda: _settings(sources_dir=str(tmp_path / "bor")),
    )
    clone_calls, fake_clone = _fake_clone()
    monkeypatch.setattr(sync_api, "clone_or_pull", fake_clone)
    _stub_probe(monkeypatch)
    fake_import = FakeImportSources(ImportSummary())
    monkeypatch.setattr(sync_api, "import_sources", fake_import)

    _login(sync_client)
    assert sync_client.post("/api/sync").status_code == 202

    body = _poll(sync_client, "failed")
    assert body["error"] == "no sources configured (git or local)"
    assert clone_calls == []  # git is never touched
    assert fake_import.sources == []


def test_db_rows_win_over_env(
    sync_client: TestClient,
    monkeypatch: pytest.MonkeyPatch,
    db: Session,
    tmp_path: Path,
    caplog: pytest.LogCaptureFixture,
) -> None:
    """Phase 35: a stored row beats ``BOR_GIT_SOURCES`` — only the DB URL
    is cloned, and the started log names the origin."""
    db_url = "https://db.example.com/managed.git"
    _seed(db, db_url)
    _stub_env(monkeypatch, "https://env.example.com/ignored.git")
    monkeypatch.setattr(
        sync_api,
        "get_settings",
        lambda: _settings(sources_dir=str(tmp_path / "bor")),
    )
    clone_calls, fake_clone = _fake_clone()
    monkeypatch.setattr(sync_api, "clone_or_pull", fake_clone)
    _stub_probe(monkeypatch)
    fake_import = FakeImportSources(ImportSummary(files=1, added=1))
    monkeypatch.setattr(sync_api, "import_sources", fake_import)
    monkeypatch.setattr(sync_api, "regenerate_overview", FakeOverview(ok=True))
    fake_folders = FakeFolderSummaries()
    monkeypatch.setattr(sync_api, "generate_folder_summaries", fake_folders)

    _login(sync_client)
    with caplog.at_level(logging.INFO, logger="app.api.sync"):
        assert sync_client.post("/api/sync").status_code == 202
        body = _poll(sync_client, "success")

    # Phase 94: the changed canned summary fired the (stubbed) folder
    # step with the run's own session + the import's LLM client.
    assert fake_folders.llms == [fake_import.llms[0]]
    assert clone_calls == [(db_url, tmp_path / "bor" / "managed")]
    assert fake_import.sources == [[tmp_path / "bor" / "managed"]]
    assert "env.example.com" not in str(body)  # the env URL never reaches the UI
    assert any("sync: started repos=1 origin=db" in r.getMessage() for r in caplog.records)


def test_env_fallback_when_table_empty(
    sync_client: TestClient,
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
    caplog: pytest.LogCaptureFixture,
) -> None:
    """Phase 35: with the table empty (the truncate fixture), the
    ``BOR_GIT_SOURCES`` list is what gets cloned — origin ``env``."""
    env_url = "https://env.example.com/fallback.git"
    _stub_env(monkeypatch, env_url)
    monkeypatch.setattr(
        sync_api,
        "get_settings",
        lambda: _settings(sources_dir=str(tmp_path / "bor")),
    )
    clone_calls, fake_clone = _fake_clone()
    monkeypatch.setattr(sync_api, "clone_or_pull", fake_clone)
    _stub_probe(monkeypatch)
    fake_import = FakeImportSources(ImportSummary(files=1, added=1))
    monkeypatch.setattr(sync_api, "import_sources", fake_import)
    monkeypatch.setattr(sync_api, "regenerate_overview", FakeOverview(ok=True))
    monkeypatch.setattr(sync_api, "generate_folder_summaries", FakeFolderSummaries())

    _login(sync_client)
    with caplog.at_level(logging.INFO, logger="app.api.sync"):
        assert sync_client.post("/api/sync").status_code == 202
        _poll(sync_client, "success")

    assert clone_calls == [(env_url, tmp_path / "bor" / "fallback")]
    assert any("sync: started repos=1 origin=env" in r.getMessage() for r in caplog.records)


# --- phase 38: the local kind (real import, host temp dirs) ---------------


def test_local_only_sync_imports_dir(
    sync_client: TestClient,
    monkeypatch: pytest.MonkeyPatch,
    db: Session,
    tmp_path: Path,
    clean_documents: None,
) -> None:
    """Local-only config: the host temp dir (one fixture ``.md``) is
    walked directly — no clone at all — and the file lands in the KB
    (``GET /api/docs`` as admin)."""
    local_dir = tmp_path / "LocalDocs"
    local_dir.mkdir()
    (local_dir / "notes.md").write_text("# Local Notes\nthe local fixture\n", encoding="utf-8")
    _seed_local(db, local_dir)
    _stub_env(monkeypatch)  # env must not matter once the table has a row
    monkeypatch.setattr(
        sync_api,
        "get_settings",
        lambda: _settings(sources_dir=str(tmp_path / "bor")),
    )
    clone_calls, fake_clone = _fake_clone()
    monkeypatch.setattr(sync_api, "clone_or_pull", fake_clone)
    _real_llm(monkeypatch)  # real import_sources, deterministic embeddings

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

    assert clone_calls == []  # nothing to clone — local is walked directly
    assert body["detail"]["added"] == 1
    assert body["detail"]["errors"] == 0
    # The local file is in the KB, sourced by the directory's basename.
    docs = sync_client.get("/api/docs").json()["documents"]
    assert [(d["source"], d["path"]) for d in docs] == [("LocalDocs", "notes.md")]


def test_mixed_git_local_sync_imports_both_and_prunes_union(
    sync_client: TestClient,
    monkeypatch: pytest.MonkeyPatch,
    db: Session,
    tmp_path: Path,
    caplog: pytest.LogCaptureFixture,
    clean_documents: None,
) -> None:
    """Mixed config: the git row is cloned, the local dir walked, and
    both are imported in one run over the single combined list. The
    started log carries the kind counts; a file deleted out of the
    local dir is pruned on the next sync (union prune) while the git
    doc survives."""
    repo_url = f"file://{tmp_path / 'repo.git'}"
    _seed(db, repo_url)
    local_dir = tmp_path / "LocalDocs"
    local_dir.mkdir()
    (local_dir / "a.md").write_text("# A\nfirst local file\n", encoding="utf-8")
    (local_dir / "b.md").write_text("# B\nsecond local file\n", encoding="utf-8")
    _seed_local(db, local_dir)
    _stub_env(monkeypatch)
    monkeypatch.setattr(
        sync_api,
        "get_settings",
        lambda: _settings(sources_dir=str(tmp_path / "bor")),
    )
    clone_calls, fake_clone = _fake_clone()
    monkeypatch.setattr(sync_api, "clone_or_pull", fake_clone)
    _real_llm(monkeypatch)

    _login(sync_client)
    with caplog.at_level(logging.INFO, logger="app.api.sync"):
        assert sync_client.post("/api/sync").status_code == 202
        body = _poll(sync_client, "success")

    # Git cloned into BOR_SOURCES_DIR, local dir walked in row order.
    assert clone_calls == [(repo_url, tmp_path / "bor" / "repo")]
    assert body["detail"]["added"] == 3
    assert any(
        "sync: started repos=2 origin=db git=1 local=1" in r.getMessage() for r in caplog.records
    )
    docs = {(d["source"], d["path"]) for d in sync_client.get("/api/docs").json()["documents"]}
    assert docs == {("repo", "notes.md"), ("LocalDocs", "a.md"), ("LocalDocs", "b.md")}

    # Union prune: delete one local file → the next sync prunes exactly
    # it; the git doc (and the surviving local file) stay.
    (local_dir / "b.md").unlink()
    assert sync_client.post("/api/sync").status_code == 202
    body = _poll(sync_client, "success")
    assert body["detail"]["pruned"] == 1
    assert body["detail"]["added"] == 0
    docs = {(d["source"], d["path"]) for d in sync_client.get("/api/docs").json()["documents"]}
    assert docs == {("repo", "notes.md"), ("LocalDocs", "a.md")}


def test_missing_local_dir_fails_loudly_and_imports_nothing(
    sync_client: TestClient, monkeypatch: pytest.MonkeyPatch, db: Session, tmp_path: Path
) -> None:
    """A local row whose directory is gone at sync time (moved/deleted
    since add-time) → ``failed`` naming the path, **zero** import
    attempts — the git row before it in row order was still cloned
    (per-row walk; a clone is not an import)."""
    repo_url = f"file://{tmp_path / 'repo.git'}"
    missing = tmp_path / "Gone"
    db.add(
        GitSource(url=repo_url, kind="git", added_at=datetime(2026, 1, 1, tzinfo=UTC))
    )
    db.add(
        GitSource(url=str(missing), kind="local", path=str(missing),
                  added_at=datetime(2026, 1, 2, tzinfo=UTC))
    )
    db.commit()
    _stub_env(monkeypatch)
    monkeypatch.setattr(
        sync_api,
        "get_settings",
        lambda: _settings(sources_dir=str(tmp_path / "bor")),
    )
    clone_calls, fake_clone = _fake_clone()
    monkeypatch.setattr(sync_api, "clone_or_pull", fake_clone)
    _stub_probe(monkeypatch)
    fake_import = FakeImportSources(ImportSummary())
    monkeypatch.setattr(sync_api, "import_sources", fake_import)
    fake_overview = FakeOverview(ok=True)
    monkeypatch.setattr(sync_api, "regenerate_overview", fake_overview)

    _login(sync_client)
    assert sync_client.post("/api/sync").status_code == 202

    body = _poll(sync_client, "failed")
    assert f"local source missing: {missing}" in body["error"]  # the path is named
    assert body["detail"] == {}
    assert clone_calls == [(repo_url, tmp_path / "bor" / "repo")]  # git row walked first
    assert fake_import.sources == []  # no partial import
    assert fake_overview.llms == []


def test_import_error_is_reported_with_credentials_masked(
    sync_client: TestClient, monkeypatch: pytest.MonkeyPatch, db: Session, tmp_path: Path
) -> None:
    repo_url = f"file://{tmp_path / 'repo.git'}"
    _seed(db, repo_url)
    _stub_env(monkeypatch)
    monkeypatch.setattr(
        sync_api,
        "get_settings",
        lambda: _settings(sources_dir=str(tmp_path / "bor")),
    )
    _, fake_clone = _fake_clone()
    monkeypatch.setattr(sync_api, "clone_or_pull", fake_clone)
    _stub_probe(monkeypatch)

    async def failing_import(
        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,  # phase 129
        on_image_progress: Callable[[int, int], None] | None = None,  # phase 152
        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,  # phase 106
    ) -> ImportSummary:
        raise EmbeddingError(
            "embeddings request to https://user:secret@aipi.reeseapps.com/v1 "
            "failed: connection refused"
        )

    monkeypatch.setattr(sync_api, "import_sources", failing_import)
    monkeypatch.setattr(sync_api, "regenerate_overview", FakeOverview(ok=True))

    _login(sync_client)
    assert sync_client.post("/api/sync").status_code == 202

    body = _poll(sync_client, "failed")
    assert "*****@aipi.reeseapps.com" in body["error"]  # credentials masked
    assert "user:secret" not in body["error"]
    assert "connection refused" in body["error"]  # the reason survives
    # Phase 98 (task 01): the failed terminal (died mid-import) clears
    # phase + current_summary; the run never reached the summary span,
    # so those counters are 0/0.
    assert body["phase"] is None
    assert body["current_summary"] is None
    assert body["summaries_done"] == 0 and body["summaries_total"] == 0


# --- phase 41: model probe (fail fast before any clone) --------------------


class _DeadEmbedder:
    """Probe fake whose embedding call dies like a dead endpoint — with a
    credential-bearing URL in the wrapped error (the sanitizer must
    mask it). ``chat`` is recorded: it must never be reached."""

    def __init__(self) -> None:
        self.settings = Settings(_env_file=None)  # pyright: ignore[reportCallIssue]
        self.embed_batches = 0
        self.chat_calls: list[list[dict[str, str]]] = []

    async def embed_one(self, text: str) -> list[float]:
        raise EmbeddingError(
            "embeddings request to https://user:secret@aipi.reeseapps.com/v1 "
            "failed: connection refused"
        )

    async def chat(self, messages: list[dict[str, str]], model: str | None = None) -> str:
        self.chat_calls.append(list(messages))
        return "pong"


def test_model_down_fails_fast_before_any_clone(
    sync_client: TestClient, monkeypatch: pytest.MonkeyPatch, db: Session, tmp_path: Path
) -> None:
    """A dead model endpoint fails the run **before any clone**: the
    model-naming error lands in the ``failed`` state verbatim (the
    sanitizer is a no-op for it), and clone + import never run."""
    repo_url = f"file://{tmp_path / 'repo.git'}"
    _seed(db, repo_url)
    _stub_env(monkeypatch)
    monkeypatch.setattr(
        sync_api,
        "get_settings",
        lambda: _settings(sources_dir=str(tmp_path / "bor")),
    )
    clone_calls, fake_clone = _fake_clone()
    monkeypatch.setattr(sync_api, "clone_or_pull", fake_clone)
    fake_import = FakeImportSources(ImportSummary())
    monkeypatch.setattr(sync_api, "import_sources", fake_import)
    message = (
        "The embedding model ('embed') is not available — "
        "check the model endpoint and retry."
    )

    async def dead_probe(llm: LLMClient) -> None:
        raise ModelUnavailableError(message)

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

    _login(sync_client)
    assert sync_client.post("/api/sync").status_code == 202

    body = _poll(sync_client, "failed")
    assert body["error"] == message  # the model is named, verbatim
    # The sanitizer must leave the model-naming message untouched.
    assert sync_api._sanitize_error(body["error"]) == body["error"]
    assert body["detail"] == {}
    assert clone_calls == []  # fail fast: before any clone
    assert fake_import.sources == []  # and before any import
    # Phase 53: a FAILED sync (model down) never bumps.
    assert current_sources_version(db) == 0


def test_probe_names_dead_embed_model_and_masks_credentials(
    sync_client: TestClient, monkeypatch: pytest.MonkeyPatch, db: Session, tmp_path: Path
) -> None:
    """The **real** probe against a stubbed dead client: the error names
    the embedding model, and the credentials wrapped from the endpoint
    URL are masked by the sync sanitizer (the reason survives)."""
    repo_url = f"file://{tmp_path / 'repo.git'}"
    _seed(db, repo_url)
    _stub_env(monkeypatch)
    monkeypatch.setattr(
        sync_api,
        "get_settings",
        lambda: _settings(sources_dir=str(tmp_path / "bor")),
    )
    dead = _DeadEmbedder()
    monkeypatch.setattr(sync_api, "LLMClient", lambda: dead)
    clone_calls, fake_clone = _fake_clone()
    monkeypatch.setattr(sync_api, "clone_or_pull", fake_clone)
    fake_import = FakeImportSources(ImportSummary())
    monkeypatch.setattr(sync_api, "import_sources", fake_import)

    _login(sync_client)
    assert sync_client.post("/api/sync").status_code == 202

    body = _poll(sync_client, "failed")
    assert "The embedding model ('embed') is not available" in body["error"]
    assert "*****@aipi.reeseapps.com" in body["error"]  # credentials masked
    assert "user:secret" not in body["error"]
    assert "connection refused" in body["error"]  # the reason survives
    assert dead.chat_calls == []  # embed died first — chat never probed
    assert clone_calls == []
    assert fake_import.sources == []


def test_probe_runs_before_source_resolution(
    sync_client: TestClient, monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
    """Ordering: the probe is the **first** pipeline step — it runs
    before ``effective_sources`` (and, transitively, before any
    clone)."""
    order: list[str] = []

    async def spy_probe(llm: LLMClient) -> None:
        order.append("probe")

    def spy_effective_sources(session: Session) -> tuple[list[GitSource], str]:
        order.append("effective_sources")
        raise GitSyncError("short-circuit after the ordering spy")

    monkeypatch.setattr(sync_api, "check_models", spy_probe)
    monkeypatch.setattr(sync_api, "effective_sources", spy_effective_sources)
    monkeypatch.setattr(
        sync_api,
        "get_settings",
        lambda: _settings(sources_dir=str(tmp_path / "bor")),
    )
    clone_calls, fake_clone = _fake_clone()
    monkeypatch.setattr(sync_api, "clone_or_pull", fake_clone)

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

    assert order == ["probe", "effective_sources"]
    assert clone_calls == []
    assert "short-circuit" in body["error"]  # the spy aborted the run


# --- phase 89: per-row ignore lists ------------------------------------------


def test_local_row_ignore_paths_excluded_from_sync(
    sync_client: TestClient,
    monkeypatch: pytest.MonkeyPatch,
    db: Session,
    tmp_path: Path,
    clean_documents: None,
) -> None:
    """Phase 89: a local row carrying ``ignore_paths`` — the button
    sync skips every matching file: the ignored file never lands in the
    KB (no document row — hence no embedding, no summary), and the
    success detail's ``files``/``added`` counts exclude it, while the
    kept file imports as usual."""
    local_dir = tmp_path / "LocalDocs"
    local_dir.mkdir()
    (local_dir / "keep.md").write_text("# Keep\nin scope\n", encoding="utf-8")
    (local_dir / "ignore").mkdir()
    (local_dir / "ignore" / "secret.md").write_text("# Secret\nignored\n",
                                                    encoding="utf-8")
    db.add(
        GitSource(
            url=str(local_dir), kind="local", path=str(local_dir),
            ignore_paths=["ignore/"],  # any spelling — the importer normalizes
        )
    )
    db.commit()
    _stub_env(monkeypatch)
    monkeypatch.setattr(
        sync_api,
        "get_settings",
        lambda: _settings(sources_dir=str(tmp_path / "bor")),
    )
    _real_llm(monkeypatch)  # real import_sources, deterministic embeddings

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

    # The ignored file is not counted: only keep.md was walked.
    assert body["detail"]["files"] == 1
    assert body["detail"]["added"] == 1
    assert body["detail"]["errors"] == 0
    # The KB holds exactly the kept file — ignore/secret.md is absent.
    docs = [(d["source"], d["path"]) for d in sync_client.get("/api/docs").json()["documents"]]
    assert docs == [("LocalDocs", "keep.md")]


def test_sync_builds_ignore_map_by_root_string_with_union(
    sync_client: TestClient,
    monkeypatch: pytest.MonkeyPatch,
    db: Session,
    tmp_path: Path,
) -> None:
    """Phase 89 wiring (fake import): the runner keys the map by the
    SAME root string the importer sees, and two rows sharing that root
    string (the sibling/repo-name edge — ``…/shared`` and
    ``…/shared.git`` clone into the same checkout dir) get the UNION
    of their lists, in row order; a row with an empty list contributes
    nothing."""
    url_a = f"file://{tmp_path / 'shared'}"
    url_b = f"{url_a}.git"  # same repo name → same checkout dir
    # Distinct added_at: the resolver orders by (added_at, id) — a
    # same-timestamp pair would tie-break on the random uuid.
    db.add(GitSource(url=url_a, kind="git", ignore_paths=["a/"],
                     added_at=datetime(2026, 1, 1, tzinfo=UTC)))
    db.add(GitSource(url=url_b, kind="git", ignore_paths=["b"],
                     added_at=datetime(2026, 1, 2, tzinfo=UTC)))
    db.commit()
    _stub_env(monkeypatch)
    monkeypatch.setattr(
        sync_api,
        "get_settings",
        lambda: _settings(sources_dir=str(tmp_path / "bor")),
    )
    _, fake_clone = _fake_clone()
    monkeypatch.setattr(sync_api, "clone_or_pull", fake_clone)
    _stub_probe(monkeypatch)
    fake_import = FakeImportSources(ImportSummary(files=1, added=1))
    monkeypatch.setattr(sync_api, "import_sources", fake_import)
    monkeypatch.setattr(sync_api, "regenerate_overview", FakeOverview(ok=True))
    monkeypatch.setattr(sync_api, "generate_folder_summaries", FakeFolderSummaries())

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

    shared = str(tmp_path / "bor" / "shared")
    # Both rows resolve to the SAME checkout (the collision itself) and
    # the map holds their union, keyed by that one root string.
    assert fake_import.sources == [[tmp_path / "bor" / "shared", tmp_path / "bor" / "shared"]]
    assert fake_import.ignore_maps == [{shared: ["a/", "b"]}]


def test_sync_without_ignore_lists_passes_empty_map(
    sync_client: TestClient,
    monkeypatch: pytest.MonkeyPatch,
    db: Session,
    tmp_path: Path,
) -> None:
    """Phase 89 regression: rows without a list → the runner passes an
    EMPTY map (the importer's byte-identical pre-phase-89 behavior)."""
    local_dir = tmp_path / "LocalDocs"
    local_dir.mkdir()
    (local_dir / "notes.md").write_text("# Notes\nplain row\n", encoding="utf-8")
    _seed_local(db, local_dir)  # no ignore_paths → []
    _stub_env(monkeypatch)
    monkeypatch.setattr(
        sync_api,
        "get_settings",
        lambda: _settings(sources_dir=str(tmp_path / "bor")),
    )
    _stub_probe(monkeypatch)
    fake_import = FakeImportSources(ImportSummary(files=1, added=1))
    monkeypatch.setattr(sync_api, "import_sources", fake_import)
    monkeypatch.setattr(sync_api, "regenerate_overview", FakeOverview(ok=True))
    monkeypatch.setattr(sync_api, "generate_folder_summaries", FakeFolderSummaries())

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

    assert fake_import.sources == [[local_dir]]
    assert fake_import.ignore_maps == [{}]  # no row carried a list


# --- phase 105: per-row hidden-folders flag --------------------------------


def test_local_row_hidden_flag_off_then_on_indexes_hidden_paths(
    sync_client: TestClient,
    monkeypatch: pytest.MonkeyPatch,
    db: Session,
    tmp_path: Path,
    clean_documents: None,
) -> None:
    """Phase 105 (A4 then A1): the SAME local row, synced first with the
    default flag (off) — the file inside the hidden folder never lands
    in the KB (no document row — hence no embedding, no summary), the
    visible file imports as usual; then the row is flipped on (direct
    model set — the PATCH round-trip is task 03's layer) and the next
    sync walks the hidden file too: it is indexed, embedded, and counted
    like any visible file."""
    local_dir = tmp_path / "LocalDocs"
    local_dir.mkdir()
    (local_dir / "visible.md").write_text("# Visible\nin scope\n", encoding="utf-8")
    (local_dir / ".hidden").mkdir()
    (local_dir / ".hidden" / "note.md").write_text("# Note\nhidden\n", encoding="utf-8")
    _seed_local(db, local_dir)  # include_hidden defaults to False (A4)
    _stub_env(monkeypatch)
    monkeypatch.setattr(
        sync_api,
        "get_settings",
        lambda: _settings(sources_dir=str(tmp_path / "bor")),
    )
    _real_llm(monkeypatch)  # real import_sources, deterministic embeddings

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

    # Flag off (A4): only the visible file is walked and indexed — the
    # hidden file has NO documents row.
    assert body["detail"]["files"] == 1
    assert body["detail"]["added"] == 1
    assert body["detail"]["errors"] == 0
    docs = {
        (d["source"], d["path"]) for d in sync_client.get("/api/docs").json()["documents"]
    }
    assert docs == {("LocalDocs", "visible.md")}

    # Flip the SAME row on — the next sync re-reads the flag per row.
    row = db.execute(select(GitSource).where(GitSource.url == str(local_dir))).scalar_one()
    row.include_hidden = True
    db.commit()

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

    # Flag on (A1): the hidden file is walked (detail.files counts it),
    # embedded, and indexed alongside the visible file.
    assert body["detail"]["files"] == 2
    assert body["detail"]["added"] == 1
    assert body["detail"]["errors"] == 0
    docs = {
        (d["source"], d["path"]) for d in sync_client.get("/api/docs").json()["documents"]
    }
    assert docs == {("LocalDocs", "visible.md"), ("LocalDocs", ".hidden/note.md")}


def test_sync_builds_include_hidden_map_by_root_string_with_or(
    sync_client: TestClient,
    monkeypatch: pytest.MonkeyPatch,
    db: Session,
    tmp_path: Path,
) -> None:
    """Phase 105 wiring (fake import): the runner keys the flag map by
    the SAME root string the importer sees, and two rows sharing that
    root string (the sibling/repo-name edge — ``…/shared`` and
    ``…/shared.git`` clone into the same checkout dir) get the OR of
    their flags — if EITHER row says "index hidden", the root does
    (the ignore-map union's boolean mirror)."""
    url_a = f"file://{tmp_path / 'shared'}"
    url_b = f"{url_a}.git"  # same repo name → same checkout dir
    # Distinct added_at: the resolver orders by (added_at, id) — a
    # same-timestamp pair would tie-break on the random uuid.
    db.add(GitSource(url=url_a, kind="git", include_hidden=False,
                     added_at=datetime(2026, 1, 1, tzinfo=UTC)))
    db.add(GitSource(url=url_b, kind="git", include_hidden=True,
                     added_at=datetime(2026, 1, 2, tzinfo=UTC)))
    db.commit()
    _stub_env(monkeypatch)
    monkeypatch.setattr(
        sync_api,
        "get_settings",
        lambda: _settings(sources_dir=str(tmp_path / "bor")),
    )
    _, fake_clone = _fake_clone()
    monkeypatch.setattr(sync_api, "clone_or_pull", fake_clone)
    _stub_probe(monkeypatch)
    fake_import = FakeImportSources(ImportSummary(files=1, added=1))
    monkeypatch.setattr(sync_api, "import_sources", fake_import)
    monkeypatch.setattr(sync_api, "regenerate_overview", FakeOverview(ok=True))
    monkeypatch.setattr(sync_api, "generate_folder_summaries", FakeFolderSummaries())

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

    shared = str(tmp_path / "bor" / "shared")
    # Both rows resolve to the SAME checkout (the collision itself) and
    # the map holds the OR of their flags, keyed by that one root
    # string — the flag-off row's False is overridden by the True.
    assert fake_import.sources == [[tmp_path / "bor" / "shared", tmp_path / "bor" / "shared"]]
    assert fake_import.include_hidden_maps == [{shared: True}]


# --- phase 134 (task 01): the cancel (the fifth state) -------------------


def test_cancel_mid_run_lands_in_cancelled_then_resyncs_clean(
    sync_client: TestClient,
    monkeypatch: pytest.MonkeyPatch,
    db: Session,
    tmp_path: Path,
    caplog: pytest.LogCaptureFixture,
) -> None:
    """Phase 134 (task 01), the API contract end to end: a real run
    started through the API is cancelled mid-run (parked inside the
    import) → ``POST /api/sync/cancel`` answers 202 ``cancel
    requested``, the status reports the new ``cancelled`` terminal
    (``error`` null, ``detail`` empty, the final file counts kept, the
    positions cleared, the PLAN §9 line logged); a cancel with no run
    in flight is 409 ``no sync is running`` (idle AND after a
    terminal); and the FOLLOWING run starts cleanly (202 — the stale
    cancel event did not bleed in) and completes to ``success`` (the
    restart after a partial cancel is clean — the importer's hash
    delta re-imports what the cancel skipped)."""
    repo_url = f"file://{tmp_path / 'repo.git'}"
    _seed(db, repo_url)
    _stub_env(monkeypatch)
    monkeypatch.setattr(
        sync_api,
        "get_settings",
        lambda: _settings(sources_dir=str(tmp_path / "bor")),
    )
    _, fake_clone = _fake_clone()
    monkeypatch.setattr(sync_api, "clone_or_pull", fake_clone)
    _stub_probe(monkeypatch)
    import_release = threading.Event()
    gated_import = _GatedImport(
        ImportSummary(files=3, added=1, updated=1, unchanged=1),
        threading.Event(),
        import_release,
    )
    monkeypatch.setattr(sync_api, "import_sources", gated_import)
    monkeypatch.setattr(sync_api, "regenerate_overview", FakeOverview(ok=True))
    monkeypatch.setattr(sync_api, "generate_folder_summaries", FakeFolderSummaries())

    _login(sync_client)
    # No run in flight (idle) → 409, the exact detail (the UI's
    # harmless race — no error copy).
    r = sync_client.post("/api/sync/cancel")
    assert r.status_code == 409
    assert r.json() == {"detail": "no sync is running"}

    assert sync_client.post("/api/sync").status_code == 202
    try:
        assert gated_import.started.wait(5.0), "the run never reached the import"
        # Parked inside the import (between files): the cancel request.
        with caplog.at_level(logging.INFO, logger="app.api.sync"):
            r = sync_client.post("/api/sync/cancel")
            assert r.status_code == 202
            assert r.json() == {"detail": "cancel requested"}
            body = _poll(sync_client, "cancelled", timeout=5.0)
    finally:
        import_release.set()

    assert body["error"] is None  # a cancel is intent, not failure
    assert body["detail"] == {}
    assert body["started_at"] is not None
    assert body["finished_at"] is not None
    assert body["current_file"] is None
    assert body["phase"] is None
    assert body["current_summary"] is None
    # The final file counts are kept (the phase-64/98 convention) ...
    assert body["files_done"] == 1 and body["files_total"] == 3
    # ... and the chunk position clears (the phase-129 convention).
    assert body["current_chunk"] == 0 and body["chunks_total"] == 0
    # PLAN §9: the ample-logging line for the converted cancel.
    assert any(
        r.getMessage()
        == "sync: cancelled by user state=cancelled files_done=1 files_total=3"
        for r in caplog.records
    )
    # The cancelled run never reached the version bump.
    assert current_sources_version(db) == 0

    # No run in flight anymore (the task is done) → 409 again.
    r = sync_client.post("/api/sync/cancel")
    assert r.status_code == 409
    assert r.json() == {"detail": "no sync is running"}

    # The restart is clean: a fresh run starts (202 — the stale event
    # did not bleed in; start_sync cleared it) and the fresh (full)
    # import completes to success.
    second_import = FakeImportSources(
        ImportSummary(files=3, added=1, updated=1, unchanged=1)
    )
    monkeypatch.setattr(sync_api, "import_sources", second_import)
    assert sync_client.post("/api/sync").status_code == 202
    body = _poll(sync_client, "success")
    assert body["error"] is None
    assert body["detail"]["added"] == 1
    assert body["detail"]["sources_version"] == 1  # only the completed run bumps
    assert current_sources_version(db) == 1
    assert len(second_import.sources) == 1
    assert second_import.prune_flags == [True]


# --- phase 135: the branch pin (real git, two-branch file:// origin) -------


def _git_available() -> bool:
    try:
        proc = subprocess.run(["git", "--version"], capture_output=True, check=False)
        return proc.returncode == 0
    except (FileNotFoundError, OSError):
        return False


#: Real ``git`` in the test environment — skipped cleanly without it
#: (the ``test_git_file_dates.py`` house pattern).
GIT = _git_available()


def _git(cwd: Path, *argv: str) -> None:
    """Run one git command for the test harness (fixture setup); a
    non-zero exit fails the fixture, not the test under test."""
    env = os.environ.copy()
    env["GIT_AUTHOR_NAME"] = "T"
    env["GIT_AUTHOR_EMAIL"] = "t@example.com"
    env["GIT_COMMITTER_NAME"] = "T"
    env["GIT_COMMITTER_EMAIL"] = "t@example.com"
    proc = subprocess.run(
        ["git", *argv], cwd=cwd, env=env, capture_output=True, text=True, check=False
    )
    assert proc.returncode == 0, f"git {' '.join(argv)} failed: {proc.stderr}"


def _two_branch_origin(tmp_path: Path) -> Path:
    """A real-git scratch origin (the phase-135 recipe): ``main`` holds
    ``a.md``, ``release`` holds ``b.md`` — different content per
    branch (a branch checkout is the branch's tree, not main's)."""
    origin = tmp_path / "origin"
    origin.mkdir()
    _git(origin, "init", "-q", "-b", "main")
    _git(origin, "config", "commit.gpgsign", "false")
    (origin / "a.md").write_text("# A\nmain branch content\n", encoding="utf-8")
    _git(origin, "add", "a.md")
    _git(origin, "commit", "-q", "-m", "main: add a.md")
    _git(origin, "checkout", "-q", "-b", "release")
    (origin / "a.md").unlink()
    (origin / "b.md").write_text("# B\nrelease branch content\n", encoding="utf-8")
    _git(origin, "add", "-A")
    _git(origin, "commit", "-q", "-m", "release: add b.md")
    _git(origin, "checkout", "-q", "main")
    return origin


@pytest.mark.skipif(not GIT, reason="git CLI not available")
def test_git_row_with_branch_button_path_clones_that_branch(
    sync_client: TestClient,
    monkeypatch: pytest.MonkeyPatch,
    db: Session,
    tmp_path: Path,
) -> None:
    """Phase 135: the BUTTON path honors the row's branch pin — a
    ``git_sources`` row with ``branch='release'`` is cloned with the
    REAL ``clone_or_pull`` (no fake in this test): the imported root
    is the RELEASE tree (file B present, file A absent)."""
    origin = _two_branch_origin(tmp_path)
    db.add(GitSource(url=f"file://{origin}", kind="git", branch="release"))
    db.commit()
    _stub_env(monkeypatch)
    monkeypatch.setattr(
        sync_api,
        "get_settings",
        lambda: _settings(sources_dir=str(tmp_path / "bor")),
    )
    _stub_probe(monkeypatch)
    fake_import = FakeImportSources(ImportSummary(files=1, added=1))
    monkeypatch.setattr(sync_api, "import_sources", fake_import)
    monkeypatch.setattr(sync_api, "regenerate_overview", FakeOverview(ok=True))
    monkeypatch.setattr(sync_api, "generate_folder_summaries", FakeFolderSummaries())

    _login(sync_client)
    assert sync_client.post("/api/sync").status_code == 202
    body = _poll(sync_client, "success")
    assert body["error"] is None

    root = tmp_path / "bor" / "origin"
    assert fake_import.sources == [[root]]
    assert (root / "b.md").is_file()  # the branch's file is what got imported
    assert not (root / "a.md").exists()


@pytest.mark.skipif(not GIT, reason="git CLI not available")
def test_git_row_branch_mismatch_button_path_fails_loud(
    sync_client: TestClient,
    monkeypatch: pytest.MonkeyPatch,
    db: Session,
    tmp_path: Path,
) -> None:
    """Phase 135 (ASSUMPTION B5): a row pinning a branch the EXISTING
    checkout does not sit on fails the button sync LOUDLY — the
    remove-and-re-add message, zero import attempts, and the checkout
    untouched (still on its branch: no silent switch)."""
    origin = _two_branch_origin(tmp_path)
    url = f"file://{origin}"
    dest = tmp_path / "bor" / "origin"
    # A pre-existing checkout on the default branch (main) — e.g. a
    # row that existed before the branch was pinned (the sync's
    # sources_dir is seeded with it before the run starts).
    clone_or_pull(url, dest)
    db.add(GitSource(url=url, kind="git", branch="release"))
    db.commit()
    _stub_env(monkeypatch)
    monkeypatch.setattr(
        sync_api,
        "get_settings",
        lambda: _settings(sources_dir=str(tmp_path / "bor")),
    )
    _stub_probe(monkeypatch)
    fake_import = FakeImportSources(ImportSummary())
    monkeypatch.setattr(sync_api, "import_sources", fake_import)
    monkeypatch.setattr(sync_api, "regenerate_overview", FakeOverview(ok=True))
    monkeypatch.setattr(sync_api, "generate_folder_summaries", FakeFolderSummaries())

    _login(sync_client)
    assert sync_client.post("/api/sync").status_code == 202

    body = _poll(sync_client, "failed")
    assert "remove and re-add the source to switch branches" in body["error"]
    assert "'main'" in body["error"] and "'release'" in body["error"]
    assert body["detail"] == {}
    assert fake_import.sources == []  # no partial import
    # Phase 53: a FAILED sync never bumps.
    assert current_sources_version(db) == 0
    # the checkout is untouched: still on main, still main's tree
    proc = subprocess.run(
        ["git", "rev-parse", "--abbrev-ref", "HEAD"],
        cwd=dest, capture_output=True, text=True, check=False,
    )
    assert proc.returncode == 0 and proc.stdout.strip() == "main"
    assert (dest / "a.md").is_file()
    assert not (dest / "b.md").exists()