"""Phase 98 task 05 E2E (Playwright, mock-only): the sync's summary
phases are visible live — endpoint + button label — and missing folder
summaries read "waiting to generate" (the "Summary pending" markers)
until a manual save or the next sync's gap-fill makes them go away.

The dedicated story suite for ``98_sync_summary_visibility`` (A16 — one
Playwright file per phase, run in isolation): the owner's report was
that while a sync runs "the number pauses for a really long time"
(after the import, the KB-overview + the per-folder summary span burn
minutes of ``lite`` calls with nothing on the wire) and that a folder
whose summary is due but missing looks "missed" in the catalog.

Run in isolation (DB must be up: ``podman compose up -d db``):

    uv run pytest tests/e2e/test_sync_summary_visibility.py -v --no-cov

MOCK-ONLY suite: ``E2E_REAL_LLM=1`` is not supported — the gate is the
deterministic mock (the phase-94 ``FOLDER_SUMMARY_MODE`` canned
``Fixture folder summary for <folder>.`` lines) plus the phase-64
``tests/e2e/slow_llm.py`` delay proxy that stretches the run past both
the ~100 ms recorder cadence and the UI's 2 s poll.

Timing fixture (the phase-64 sizing rationale, re-sized for the phase
labels): ``SLOW_DELAY_S = 1.5`` — ONE LLM call outlives the recorder's
~100 ms cadence by ~15×, so every phase span (the overview's one call
≈ 1.5 s, the summaries' three ≈ 4.5 s) is densely sampled by the
recorder AND outlives the UI's 2 s status poll (any window ≥ the poll
period catches a tick — the rendered-label assertions are
deterministic, not races). The whole changed sync is ≈ 10 LLM calls
(2 model probes + 4 file embeds + 1 overview + 3 folder summaries)
≈ 15–18 s — well inside the generous settle budgets.

The suite runs TWO module apps (the phase-79 full-inventory pattern —
each binds its own port) over the shared E2E Postgres:

* the SLOW app — ``BOR_LLM_BASE_URL`` = the slow proxy in front of the
  mock: tests 1–2 (the phase machine at the endpoint + the phase-aware
  button label);
* the FAST app — ``BOR_LLM_BASE_URL`` = the mock directly (it stays on
  its port for the fast tests): test 3 (the pending markers + the
  gap-fill recovery — the canned summaries are byte-stable).

KB fixture — a host temp dir (``tmp_path_factory``; the app runs on
the same host) registered as a ``kind=local`` source (the
``test_local_directory_sources.py`` / phase-94 API-registration
pattern; no git anywhere): ``syncsum/`` with TWO ≥ 2-doc folders —
``alpha/`` (2 docs) and ``bravo/`` (2 docs), no root-level files. The
generator's candidate set is therefore exactly 3: the source root
(``folder_path ""``) + the two folders (the recursive-subtree ≥ 2 rule).

Autouse cleanup (the phase-96 pattern): before each test wait for no
running sync on EITHER app, then truncate the shared registry + KB
tables (documents/chunks/query_log/steering_notes/kb_overview/
git_sources/folder_summaries) — each test seeds its own source and
runs its own sync; after each test the same truncate.

Test → observable mapping (Playwright Mapping Rule):
1. ``test_sync_status_reports_the_summary_phases`` (endpoint — the
   slow leg, the phase-64 tight-poll recorder): some running tick has
   ``phase "import"`` (with the live ``current_file`` + counts the
   import always carried), some have ``phase "overview"``, some have
   ``phase "summaries"`` with ``summaries_total == 3``, non-null
   ``current_summary`` (starting with the source name — one tick the
   BARE source name, the source-root call) and ``summaries_done``
   climbing 1 → 2 → 3; every summaries tick keeps
   ``files_done == files_total`` (the import is finished — the
   reported pause is now labeled); the terminal has ``phase``/
   ``current_summary`` null but keeps ``summaries_done ==
   summaries_total == 3`` (the keep-final-counts convention).
2. ``test_sync_button_names_the_summary_phase`` (UI — the slow leg):
   a fresh admin page on ``/sources.html`` clicks **Sync sources**;
   the page's OWN 2 s poll renders ``#sync-label`` matching
   ``Summarizing folders… [folder] (n/3)`` (the asserted text is the
   RENDERED DOM label, with the button ``title`` riding the same
   untruncated value — A4); on settle the button reads the terminal
   ``Synced HH:MM`` (the phase-32 contract) and ``#sync-result``
   carries the counts line.
3. ``test_missing_folder_summaries_read_as_pending_and_self_heal``
   (tree — the fast mock leg): after a successful sync (every
   candidate stores its canned line) NO marker is anywhere; DELETE
   one folder's row AND the source-root row directly (the phase-96
   row-deletion pattern) + re-fetch (the phase-77 re-show refresh) →
   the two affected rows show ``Summary pending`` (+ the D4 title,
   the D4 class, the always-present Edit) while the intact folder
   shows its stored line and NO marker; the source level AND the
   affected folder's level block show the D4 pending note; saving a
   manual description from the row's Edit clears the marker IN PLACE
   (no reload); a second UNCHANGED sync's gap-fill regenerates the
   OTHER deleted row (the source root) — its marker is gone, its
   cell/level carry the deterministic mock line, and the manual row
   is untouched (``only_missing`` + ``manually_edited``).
"""
from __future__ import annotations

import json
import os
import re
import subprocess
import sys
import threading
import time
from collections.abc import Iterator
from datetime import datetime
from pathlib import Path
from typing import Any

import httpx
import pytest
from playwright.sync_api import Page, expect
from sqlalchemy import select, text

from app.config import Settings as _Settings
from app.db import SessionLocal
from app.models import FolderSummary
from e2e.auth_helpers import login
from e2e.conftest import (
    ADMIN_PASSWORD,
    MOCK_PORT,
    SESSION_SECRET,
    USE_REAL_LLM,
    _wait_http,
)

REPO = Path(__file__).resolve().parents[2]

# Phase 79 (task 04, full inventory): the conftest session app owns its
# port in a combined run — this module's apps bind their own ports
# instead (a same-port second uvicorn dies on bind and would drive the
# wrong server). Env-overridable.
SLOW_APP_PORT = int(os.environ.get("E2E_APP_PORT_SYNCSUM_SLOW", "8142"))
FAST_APP_PORT = int(os.environ.get("E2E_APP_PORT_SYNCSUM_FAST", "8143"))
SLOW_APP_URL = f"http://127.0.0.1:{SLOW_APP_PORT}"
FAST_APP_URL = f"http://127.0.0.1:{FAST_APP_PORT}"

#: The slow-LLM proxy's port (the conftest's mock LLM stays on
#: MOCK_PORT — the FAST app points at it directly for test 3).
SLOW_PORT = int(os.environ.get("E2E_SLOW_LLM_PORT_SYNCSUM", "8904"))
SLOW_URL = f"http://127.0.0.1:{SLOW_PORT}"

#: Per-LLM-request delay on the proxy — sized so ONE LLM call outlives
#: the recorder's ~100 ms cadence by ~15× (the phase-64 rationale):
#: the overview span (one call ≈ 1.5 s) and the summaries span (three
#: calls ≈ 4.5 s) are densely sampled by the recorder AND outlive the
#: UI's 2 s poll period (any window ≥ the period catches a tick — the
#: rendered-label assertion is deterministic). A changed sync is
#: ≈ 10 calls ≈ 15–18 s (see the module docstring).
SLOW_DELAY_S = "1.5"

# --------------------------------------------------------------------------
# Fixture constants (deterministic, token-controlled)
# --------------------------------------------------------------------------

#: The local source — the temp directory's basename (``kind=local`` →
#: the directory's basename is the source name, phase 38).
SOURCE = "syncsum"

#: The two ≥ 2-doc folders (path order — the tree rows' order).
FOLDER_A = "alpha"
FOLDER_B = "bravo"
FOLDERS = (FOLDER_A, FOLDER_B)

N_FILES = 4  # two folders × 2 docs
N_CANDIDATES = 3  # the source root + the two folders (the ≥ 2-doc rule)

#: The mock's byte-stable ``FOLDER_SUMMARY_MODE`` lines for this fixture
#: (the phase-94 template — the label is the ``FOLDER_HEADER_PREFIX``
#: tail: ``<source>`` for the root, ``<source>/<folder>`` for a folder).
SUM_ROOT = f"Fixture folder summary for {SOURCE}."
SUM_ALPHA = f"Fixture folder summary for {SOURCE}/{FOLDER_A}."
SUM_BRavo = f"Fixture folder summary for {SOURCE}/{FOLDER_B}."

#: The D4 "Summary pending" marker copy (phase 98 task 04 — verbatim
#: from the D4 decision): the row-cell text, the row-cell ``title``,
#: and the level-block pending note.
PENDING_COPY = "Summary pending"
PENDING_TITLE = "No stored description yet — the next sync will generate one."
PENDING_NOTE = (
    "No description stored yet — the next sync will generate one. "
    "(You can write one yourself.)"
)

#: The hand-written description test 3 saves on the affected folder —
#: a distinctive sentence no part of the fixture or the canned
#: template contains (the round-trip assertions can never pass against
#: the old text, the phase-97 pattern).
MANUAL_BRavo = (
    "Owner note: bravo holds the bravo-a and bravo-b fixture files — "
    "written by hand after the row was deleted. (RESE-SYNCSUM-01)"
)

#: "Synced HH:MM" — the local-time last-result label (sources.js's
#: fmtSyncTime), any hour/minute (the test_sync_button.py pattern).
SYNCED_LABEL = re.compile(r"Synced \d{1,2}:\d{2}$")

#: Generous settle budgets: a changed sync against the slowed LLM is
#: ≈ 15–18 s; the UI's 2 s poll settles at most one tick after the
#: terminal state lands.
SETTLE_TIMEOUT_MS = 90_000


def _md(title: str, body: str) -> str:
    return f"# {title}\n\n{body}\n"


# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------


@pytest.fixture(scope="module")
def syncsum_dir(tmp_path_factory: pytest.TempPathFactory) -> Path:
    """The one-source temp tree (see the module docstring): the app
    servers run on the same host, so the paths are visible to them.
    The directory NAME is the source name (``kind=local``, phase 38)."""
    root = tmp_path_factory.mktemp("bor_synccom")
    src = root / SOURCE
    for folder in FOLDERS:
        (src / folder).mkdir(parents=True)
        for letter in ("a", "b"):
            (src / folder / f"{folder}-{letter}.md").write_text(
                _md(
                    f"Synccom {folder.title()} {letter.upper()}",
                    f"Synccom {folder} fixture note {letter.upper()}: covers "
                    f"topic {letter.upper()} of the {SOURCE} source tree.",
                ),
                encoding="utf-8",
            )
    assert len(list(src.rglob("*.md"))) == N_FILES
    return src


def _app_env(base_url: str, tmp_path_factory: pytest.TempPathFactory) -> dict[str, str]:
    """The per-module app env (the conftest pattern, cf.
    ``test_oneshot_llm_retry.py`` / ``test_kb_tree.py``): the
    ``base_url`` picks the LLM (slow proxy vs. the mock directly),
    ``BOR_GIT_SOURCES`` forced empty (the sync source is this suite's
    own DB-registered local row), and the leak-guarded code defaults."""
    env = dict(os.environ)
    env.pop("DEBUGPY", None)
    env["BOR_ENVIRONMENT"] = "e2e"
    env["BOR_STATIC_DIR"] = str(REPO / "frontend")
    env["BOR_LLM_BASE_URL"] = (
        "https://aipi.reeseapps.com/v1" if USE_REAL_LLM else base_url
    )
    # Mock-calibrated threshold (conftest pattern): this suite never
    # asks the chat model anything — the gate is never on a path.
    env["BOR_RELEVANCE_THRESHOLD"] = "0.30"
    # Phase 67: instant retry waits + the code-default budget (the
    # conftest leak-guard pattern).
    env["BOR_LLM_RETRY_DELAY"] = "0"
    env["BOR_LLM_RETRIES"] = str(_Settings.model_fields["llm_retries"].default)
    env.setdefault(
        "BOR_DATABASE_URL",
        "postgresql+psycopg://reese:reese@localhost:5432/brain_of_reese",
    )
    # Phase 16: admin auth must be set or create_app() refuses to boot.
    env["BOR_ADMIN_PASSWORD"] = ADMIN_PASSWORD
    env["BOR_SESSION_SECRET"] = SESSION_SECRET
    # The repo's .env file carries the owner's BOR_GIT_SOURCES (the app
    # reads it from cwd) — override it with an EMPTY value (the env var
    # beats the .env file): the registry must hold EXACTLY the local
    # directory this suite registers.
    env["BOR_GIT_SOURCES"] = ""
    # Leak guards (conftest pattern): an operator's local (gitignored)
    # .env cannot leak corpus-specific settings into the app under test.
    env["BOR_DOCS_REPO"] = ""
    env["BOR_SUGGESTIONS"] = json.dumps(
        _Settings.model_fields["suggestions"].default
    )
    env["BOR_INPUT_PLACEHOLDER"] = _Settings.model_fields["input_placeholder"].default
    env["BOR_FOOTER_TEXT"] = _Settings.model_fields["footer_text"].default
    env["BOR_SOURCES_DIR"] = str(tmp_path_factory.mktemp("bor_checkouts"))
    return env


@pytest.fixture(scope="module")
def slow_llm(mock_llm: int) -> Iterator[int]:
    """The delay-injecting reverse proxy in front of the mock LLM
    (``tests/e2e/slow_llm.py``) — the timing fixture for tests 1–2:
    the phase-machine ticks and the rendered phase label need the run
    to outlive both the ~100 ms recorder cadence and the UI's 2 s
    poll (see ``SLOW_DELAY_S``)."""
    env = dict(os.environ)
    env.pop("DEBUGPY", None)
    env["SLOW_LLM_DELAY_S"] = SLOW_DELAY_S
    env["E2E_MOCK_PORT"] = str(MOCK_PORT)
    proc = subprocess.Popen(
        [sys.executable, "-m", "uvicorn", "tests.e2e.slow_llm:app",
         "--host", "127.0.0.1", "--port", str(SLOW_PORT), "--log-level", "warning"],
        cwd=REPO,
        env=env,
    )
    try:
        _wait_http(f"{SLOW_URL}/v1/models")
        yield SLOW_PORT
    finally:
        proc.terminate()
        try:
            proc.wait(timeout=10)
        except subprocess.TimeoutExpired:
            proc.kill()


@pytest.fixture(scope="module")
def app_server(
    mock_llm: int,
    slow_llm: int,
    syncsum_dir: Path,
    tmp_path_factory: pytest.TempPathFactory,
) -> Iterator[str]:
    """The SLOW app under test (tests 1–2): ``BOR_LLM_BASE_URL`` is the
    slow proxy in front of the mock (the timing fixture — see the
    module docstring). ``syncsum_dir`` is a dependency only for the
    fixture ordering (the temp tree exists before the first test)."""
    env = _app_env(f"{SLOW_URL}/v1", tmp_path_factory)
    proc = subprocess.Popen(
        [sys.executable, "-m", "uvicorn", "app.main:app",
         "--host", "127.0.0.1", "--port", str(SLOW_APP_PORT), "--log-level", "warning"],
        cwd=REPO,
        env=env,
    )
    try:
        _wait_http(f"{SLOW_APP_URL}/api/health")
        yield SLOW_APP_URL
    finally:
        proc.terminate()
        try:
            proc.wait(timeout=10)
        except subprocess.TimeoutExpired:
            proc.kill()


@pytest.fixture(scope="module")
def app_url(app_server: str) -> str:
    return app_server


@pytest.fixture(scope="module")
def fast_app_server(
    mock_llm: int,
    syncsum_dir: Path,
    tmp_path_factory: pytest.TempPathFactory,
) -> Iterator[str]:
    """The FAST app under test (test 3): ``BOR_LLM_BASE_URL`` is the
    mock DIRECTLY (it stays on its port for the fast tests — the
    canned ``FOLDER_SUMMARY_MODE`` lines are byte-stable there, and
    the pending-marker / gap-fill assertions key on that exact text)."""
    env = _app_env(f"http://127.0.0.1:{mock_llm}/v1", tmp_path_factory)
    proc = subprocess.Popen(
        [sys.executable, "-m", "uvicorn", "app.main:app",
         "--host", "127.0.0.1", "--port", str(FAST_APP_PORT), "--log-level", "warning"],
        cwd=REPO,
        env=env,
    )
    try:
        _wait_http(f"{FAST_APP_URL}/api/health")
        yield FAST_APP_URL
    finally:
        proc.terminate()
        try:
            proc.wait(timeout=10)
        except subprocess.TimeoutExpired:
            proc.kill()


@pytest.fixture(scope="module")
def fast_app_url(fast_app_server: str) -> str:
    return fast_app_server


def _truncate_all() -> None:
    """Fresh registry + KB (the E2E isolation pattern): the E2E suites
    share one Postgres, so a leftover git_sources row or document would
    pollute the top-level rows, the candidate count, and the counts
    the syncs and the tree assertions pin exactly."""
    with SessionLocal() as db:
        db.execute(
            text(
                "TRUNCATE chunks, documents, query_log, steering_notes, "
                "kb_overview, git_sources, folder_summaries"
            )
        )
        db.commit()  # without the commit the TRUNCATE rolls back (the house pattern)


def _wait_sync_settled(base_url: str, timeout_s: float = 180.0) -> None:
    """No background sync may leak across tests (the run state lives in
    the app's memory, and a still-running run would keep importing into
    the NEXT test's truncated KB): wait for the app's sync status to
    be non-running BEFORE the truncate. Own admin session (the
    endpoint is admin-only) — usually a no-op: the tests settle only
    after their run's terminal state."""
    with httpx.Client(base_url=base_url, timeout=5.0) as client:
        r = client.post("/api/login", json={"password": ADMIN_PASSWORD})
        assert r.status_code == 204, r.text
        deadline = time.monotonic() + timeout_s
        while time.monotonic() < deadline:
            r = client.get("/api/sync/status")
            assert r.status_code == 200, r.text
            if r.json()["state"] != "running":
                return
            time.sleep(0.2)
    raise AssertionError(f"a sync was still running at a test boundary: {base_url}")


@pytest.fixture(autouse=True)
def _clean(app_url: str, fast_app_url: str, db_ready: None) -> Iterator[None]:
    """Per-test KB isolation (the phase-96 cleanup pattern): the
    shared E2E Postgres is truncated before AND after every test —
    each test seeds its own source and runs its own sync(s), so the
    KB rows + folder_summaries rows of the temp source are exactly
    this test's own. The pre-wait keeps a leaked run from the
    previous test (module apps are shared) from racing the truncate."""
    for base in (app_url, fast_app_url):
        _wait_sync_settled(base)
    _truncate_all()
    yield
    _truncate_all()


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------


def _register_source(base_url: str, path: str) -> None:
    """Register the temp directory through the authenticated API (the
    ``test_local_directory_sources.py`` / phase-94 pattern — a plain
    directory is an API/DB-only operation)."""
    with httpx.Client(base_url=base_url, timeout=30.0) as client:
        r = client.post("/api/login", json={"password": ADMIN_PASSWORD})
        assert r.status_code == 204, r.text
        r = client.post(
            "/api/git-sources", json={"kind": "local", "path": path}
        )
        assert r.status_code == 201, r.text


def _run_sync(base_url: str, timeout_s: float = 120.0) -> dict[str, Any]:
    """Login + ``POST /api/sync`` + poll the status endpoint until the
    run reaches a terminal state (the ``test_local_directory_sources``
    pattern, over plain httpx)."""
    with httpx.Client(base_url=base_url, timeout=30.0) as client:
        r = client.post("/api/login", json={"password": ADMIN_PASSWORD})
        assert r.status_code == 204, r.text
        r = client.post("/api/sync")
        assert r.status_code == 202, r.text
        deadline = time.monotonic() + timeout_s
        body: dict[str, Any] = {}
        while time.monotonic() < deadline:
            r = client.get("/api/sync/status")
            assert r.status_code == 200, r.text
            body = r.json()
            if body["state"] in ("success", "failed"):
                return body
            time.sleep(0.5)
    raise AssertionError(f"sync did not reach a terminal state: {body}")


def _folder_rows() -> dict[str, tuple[str, bool]]:
    """The source's stored folder summaries
    ``{folder_path: (summary, manually_edited)}`` (``""`` = the source
    root) — the test process's direct DB access (the E2E's other
    established lens)."""
    with SessionLocal() as db:
        rows = db.execute(
            select(
                FolderSummary.folder_path,
                FolderSummary.summary,
                FolderSummary.manually_edited,
            ).where(FolderSummary.source == SOURCE)
        ).all()
    return {
        folder: (summary, manually)
        for folder, summary, manually in rows
    }


def _delete_folder_rows(pairs: list[tuple[str, str]]) -> None:
    """DELETE stored ``folder_summaries`` rows directly (the phase-96
    row-deletion pattern — simulating a historical fail-soft miss):
    the rows are gone from the app's back without any KB change."""
    with SessionLocal() as db:
        for source, folder in pairs:
            db.execute(
                text(
                    "DELETE FROM folder_summaries "
                    "WHERE source = :s AND folder_path = :f"
                ),
                {"s": source, "f": folder},
            )
        db.commit()


class _TickRecorder:
    """The deterministic layer, concurrent with the test's own client.

    A daemon thread that tight-polls (~100 ms cadence)
    ``GET /api/sync/status`` with its OWN admin session (``httpx`` —
    the test's client drives itself in the meantime; the Playwright
    sync API is not thread-safe, so the thread never touches it),
    recording every tick from the first poll: the idle prelude, the
    running ticks, and the terminal body. The thread only READS the
    same endpoint the UI's 2 s poll reads — it starts no jobs and
    cannot skew the run. (The phase-64 recorder pattern.)

    The stale-terminal guard (the module apps are shared — a previous
    test's terminal lives in the app's memory): a terminal is
    accepted only after a RUNNING tick was observed (this suite's runs
    outlive the 100 ms cadence by ~15× — ``require_running`` is
    always on here)."""

    def __init__(self, app_url: str, path: str) -> None:
        self._url = f"{app_url}{path}"
        self._login_url = f"{app_url}/api/login"
        self._t0 = time.time()
        self._ticks: list[dict[str, Any]] = []
        self._terminal: dict[str, Any] | None = None
        self._stop = threading.Event()
        self._thread: threading.Thread | None = None

    def _terminal_is_this_run(self, body: dict[str, Any], saw_running: bool) -> bool:
        if saw_running:
            return True
        started = body.get("started_at")
        if not started:
            return False
        try:
            started_at = datetime.fromisoformat(str(started)).timestamp()
        except ValueError:
            return False
        return started_at >= self._t0 - 2.0  # tolerance for the pre-submit gap

    def start(self) -> None:
        def run() -> None:
            with httpx.Client(timeout=5.0) as client:
                # Its own admin session — the recorder's reads must not
                # depend on (or disturb) the test's client's cookie.
                client.post(self._login_url, json={"password": ADMIN_PASSWORD})
                saw_running = False
                while not self._stop.is_set():
                    try:
                        r = client.get(self._url)
                        if r.status_code == 200:
                            body = r.json()
                            self._ticks.append(body)
                            if body["state"] == "running":
                                saw_running = True
                            elif (
                                body["state"] in ("success", "failed")
                                and self._terminal_is_this_run(body, saw_running)
                            ):
                                self._terminal = body
                                return
                    except Exception:  # noqa: BLE001 — blip: retry next tick
                        pass
                    self._stop.wait(0.1)

        self._thread = threading.Thread(target=run, daemon=True)
        self._thread.start()

    def stop(self, timeout_s: float = 120.0) -> dict[str, Any]:
        """Join until a terminal tick is recorded (or fail the test)."""
        deadline = time.monotonic() + timeout_s
        while time.monotonic() < deadline:
            if self._terminal is not None:
                break
            time.sleep(0.05)
        self._stop.set()
        if self._thread is not None:
            self._thread.join(timeout=5)
        assert self._terminal is not None, (
            "the recorder saw no terminal state "
            f"(last ticks: {self._ticks[-3:] if self._ticks else 'none'})"
        )
        return self._terminal

    @property
    def running_ticks(self) -> list[dict[str, Any]]:
        return [t for t in self._ticks if t["state"] == "running"]


def _wait_top_level(page: Page) -> None:
    """The admin boot / re-fetch has rendered the top level: the single
    source row is in the folders table (the tree's fetch settled) and
    the breadcrumb is hidden (the top level is home)."""
    expect(page.locator("#folders-tbody tr")).to_have_count(1, timeout=30_000)
    expect(page.locator("#kb-crumb")).to_be_hidden()


def _drill(page: Page, *names: str) -> None:
    """Drill one level at a time (client-side — no fetch, no URL
    change): each name is the EXACT text of the source/folder link at
    the current level (the row builders' link text: the source name,
    or the folder's last path segment — the phase-97 pattern)."""
    for name in names:
        page.click(f'#folders-tbody a.folder-link:text-is("{name}")')


# ---------------------------------------------------------------------------
# 1. Endpoint — the phase machine: import keeps the file label, the
#    overview is named, the folder span reports the folder + a
#    done/total that climbs to the candidate count
# ---------------------------------------------------------------------------


def test_sync_status_reports_the_summary_phases(
    app_url: str, syncsum_dir: Path, db_ready: None
) -> None:
    """A KB-changing sync of the temp source (candidates: the source
    root + the two folders = 3) is tight-polled by the ~100 ms
    recorder from the 202 to the terminal (the phase-64 pattern): the
    recorded ticks show the D1 phase machine end to end — the
    model-check prelude keeps ``phase`` null, the import keeps its
    byte-identical live-file fields with ``phase "import"``, the
    overview span is named, and the folder span carries
    ``current_summary`` + ``summaries_done``/``summaries_total``
    (climbing 1 → 2 → 3) with ``files_done == files_total`` (the
    import is finished — the pause the user reported) — while the
    terminal clears ``phase``/``current_summary`` but KEEPS the final
    summary counts (the phase-64 keep-final-counts convention)."""
    _register_source(app_url, str(syncsum_dir))

    recorder = _TickRecorder(app_url, "/api/sync/status")
    recorder.start()
    with httpx.Client(base_url=app_url, timeout=30.0) as client:
        r = client.post("/api/login", json={"password": ADMIN_PASSWORD})
        assert r.status_code == 204, r.text
        r = client.post("/api/sync")
        assert r.status_code == 202, r.text
    terminal = recorder.stop(timeout_s=120)

    # --- the terminal (D1: clear the phase + folder, keep the final
    # summary counts; the phase-64 keep-final-counts convention extends
    # to the file counts) ---
    assert terminal["state"] == "success", terminal
    detail = terminal["detail"]
    assert detail["added"] == N_FILES, detail
    assert terminal["current_file"] is None
    assert terminal["files_done"] == N_FILES
    assert terminal["files_total"] == N_FILES
    assert terminal["phase"] is None, terminal
    assert terminal["current_summary"] is None, terminal
    assert terminal["summaries_done"] == N_CANDIDATES, terminal
    assert terminal["summaries_total"] == N_CANDIDATES, terminal

    running = recorder.running_ticks
    assert running, "the recorder caught no running tick"

    # --- the prelude (model check): ``phase`` null — the bare
    # "Syncing…" label stays (the phase-64 pins hold). Two probes ×
    # 1.5 s under the slow proxy: the recorder cannot miss it. ---
    prelude = [t for t in running if t["phase"] is None]
    assert prelude, f"no phase-less prelude tick: {running[:6]}"
    assert all(t["current_file"] is None for t in prelude), prelude

    # --- the import phase: the byte-identical file label + counts
    # (phase 64), now carrying ``phase "import"`` (D1) ---
    imp = [t for t in running if t["phase"] == "import"]
    assert imp, f"no import-phase tick: {[t['phase'] for t in running]}"
    with_file = [t for t in imp if t["current_file"]]
    assert with_file, f"no import tick carried a current_file: {imp[:6]}"
    assert all(
        t["current_file"].startswith(f"{SOURCE}/") for t in with_file
    ), with_file
    assert all(t["files_total"] == N_FILES for t in with_file), with_file
    dones = [t["files_done"] for t in with_file]
    assert dones == sorted(dones), f"files_done not monotonic: {dones}"
    assert dones[0] >= 1 and max(dones) == N_FILES, dones

    # --- the overview phase: named (D1) ---
    ov = [t for t in running if t["phase"] == "overview"]
    assert ov, f"no overview-phase tick: {[t['phase'] for t in running]}"

    # --- the summaries phase (D1/D5): the folder being summarized +
    # the done/total through the long span ---
    sm = [t for t in running if t["phase"] == "summaries"]
    assert sm, f"no summaries-phase tick: {[t['phase'] for t in running]}"
    # EVERY summaries tick: the import is finished — the file count
    # sits still (the reported pause is now labeled).
    for t in sm:
        assert t["files_done"] == t["files_total"] == N_FILES, t
    # The steady ticks (after the generator's loop starts): total is
    # the candidate count and the folder part is non-null, starting
    # with the source name (D1's shape).
    steady = [t for t in sm if t["summaries_total"] == N_CANDIDATES]
    assert steady, f"no steady summaries tick (total {N_CANDIDATES}): {sm[:4]}"
    for t in steady:
        assert t["current_summary"], t
        assert t["current_summary"].startswith(SOURCE), t
    # The per-done positions: the sorted candidate order is
    # (root, alpha, bravo) — the bare source name is the ROOT call,
    # the folders ride ``source/folder``. Under the 15× sizing each
    # position is held ≈ 1.5 s, so the recorded progression is exactly
    # 1 → 2 → 3 (monotonic, every value present, never revisited).
    by_done: dict[int, list[str]] = {}
    for t in steady:
        by_done.setdefault(t["summaries_done"], []).append(t["current_summary"])
    assert sorted(by_done) == [1, 2, 3], f"summaries_done progression: {sorted(by_done)}"
    assert set(by_done[1]) == {SOURCE}, by_done[1]  # the bare source name — the root
    assert set(by_done[2]) == {f"{SOURCE}/{FOLDER_A}"}, by_done[2]
    assert set(by_done[3]) == {f"{SOURCE}/{FOLDER_B}"}, by_done[3]

    # --- the phase machine's ORDER: null → import → overview →
    # summaries (summaries is the last phase — the run goes straight
    # to the bump/terminal after it) ---
    phased = [t for t in running if t["phase"] is not None]
    assert phased[0]["phase"] == "import", phased[0]
    first_sm = next(
        i for i, t in enumerate(phased) if t["phase"] == "summaries"
    )
    assert any(t["phase"] == "overview" for t in phased[:first_sm])
    assert all(t["phase"] == "summaries" for t in phased[first_sm:])


# ---------------------------------------------------------------------------
# 2. UI — the sync button names the phase: the page's own 2 s poll
#    renders the summaries label (folder + counts), then the phase-32
#    terminal settle
# ---------------------------------------------------------------------------


def test_sync_button_names_the_summary_phase(
    page: Page, app_url: str, syncsum_dir: Path, db_ready: None
) -> None:
    """A fresh admin page on ``/sources.html`` clicks **Sync sources**;
    the page's OWN 2 s poll renders ``#sync-label`` as
    ``Summarizing folders… [source/folder] (n/3)`` (asserted on the
    RENDERED text — the label is built by the page JS — with the
    button ``title`` carrying the same untruncated value, A4); on
    settle the button shows the terminal ``Synced HH:MM`` (the
    phase-32 contract) and ``#sync-result`` carries the counts line.
    The prelude/import labels stay the phase-64 shape (the endpoint
    test pins those phases; the UI pins the new summaries label)."""
    page.set_default_timeout(30_000)
    _register_source(app_url, str(syncsum_dir))
    login(page, app_url)  # lands on /sources.html (the button's home)

    btn = page.locator("#sync-btn")
    expect(btn).to_be_visible(timeout=30_000)
    # The button is settled and retry-ready at boot: idle ("Sync
    # sources") — or the RE-ATTACHED terminal from the previous
    # test's sync (the module app is shared, its in-memory sync state
    # survives the test boundary, and the phase-32 boot re-attach
    # renders the last result). Waiting on the settled label also
    # proves the boot re-attach COMPLETED (no later re-render can
    # clobber the running state) — the phase-64 pattern.
    expect(page.locator("#sync-label")).to_have_text(
        re.compile(r"^(Sync sources|Synced \d{1,2}:\d{2})$")
    )
    expect(btn).to_be_enabled()

    btn.click()

    # The click's immediate state (the 202 moment — the bare prefix,
    # the phase-64 contract): disabled, aria-busy, spinning icon, no
    # error…
    expect(btn).to_be_disabled()
    expect(btn).to_have_attribute("aria-busy", "true")
    expect(btn.locator(".sync-icon")).to_have_class(re.compile(r"\bis-spinning\b"))
    expect(page.locator("#sync-label")).to_have_text("Syncing…")
    expect(page.locator("#sync-error-banner")).to_be_hidden()

    # UI layer (D2): the page's own 2 s poll names the summaries phase
    # — the RENDERED label matches the D2 shape (the folder part is
    # optional — the phase's first poll may precede the first hook)
    # with the (n/3) counts, and the button ``title`` carries the same
    # untruncated value (A4). ``wait_for_function`` checks BOTH in ONE
    # in-browser evaluation: a poll tick updates label + title together
    # in a single JS task, so the atomic check can never straddle one
    # (the JS regex is the D2 shape, unit-pinned in
    # tests/unit/test_frontend_sync_upload.py).
    page.wait_for_function(
        """() => {
            const label = document.querySelector('#sync-label');
            const btn = document.querySelector('#sync-btn');
            if (!label || !btn) return false;
            const text = label.textContent;
            const re = /^Summarizing folders…(?: \\S+)? \\(\\d+\\/3\\)$/;
            return re.test(text) && btn.getAttribute('title') === text;
        }""",
        timeout=SETTLE_TIMEOUT_MS,
    )

    # The success settle (the phase-32 contract, preserved): "Synced
    # HH:MM" + the counts result line, the button re-enabled, no
    # error — the run's whole span (prelude + import + overview +
    # summaries) was labeled along the way (test 1 pins the rest).
    expect(page.locator("#sync-label")).to_have_text(
        SYNCED_LABEL, timeout=60_000
    )
    expect(btn).to_be_enabled()
    expect(btn).not_to_have_attribute("aria-busy")
    expect(page.locator("#sync-result")).to_have_text(f"{N_FILES} added")


# ---------------------------------------------------------------------------
# 3. Tree — the pending markers: missing rows read "waiting", a manual
#    save clears the marker in place, and the next unchanged sync's
#    gap-fill heals the other deleted row
# ---------------------------------------------------------------------------


def test_missing_folder_summaries_read_as_pending_and_self_heal(
    page: Page, fast_app_url: str, syncsum_dir: Path, db_ready: None
) -> None:
    """After a successful (fast mock) sync — every candidate stores
    its deterministic canned line — NO ``Summary pending`` is
    anywhere. DELETE one folder's row AND the source-root row directly
    (the phase-96 pattern) + re-fetch (the phase-77 re-show refresh) →
    the two affected rows show ``Summary pending`` (+ the D4 title,
    the D4 class, the always-present Edit) while the intact folder
    shows its stored line and NO marker; the source level AND the
    affected folder's level block show the D4 pending note. Saving a
    manual description from the row's Edit clears the marker IN PLACE
    (the cell shows the saved text — no reload). A second UNCHANGED
    sync's gap-fill then regenerates the OTHER deleted row (the
    source root): its marker is gone, its cell/level carry the
    deterministic mock line — and the manual row is untouched
    (``only_missing`` + ``manually_edited``)."""
    page.set_default_timeout(30_000)
    login(page, fast_app_url)  # lands on /sources.html (the RAG view)

    # The changed sync under the fast mock: every candidate (root +
    # the two folders) stores its canned line; the terminal keeps the
    # full-regeneration summary counts.
    _register_source(fast_app_url, str(syncsum_dir))
    body = _run_sync(fast_app_url)
    assert body["state"] == "success", body
    detail = body["detail"]
    assert detail["added"] == N_FILES, detail
    assert detail["updated"] == 0 and detail["pruned"] == 0, detail
    assert detail["overview"] is True, detail
    assert body["summaries_done"] == N_CANDIDATES, body
    assert body["summaries_total"] == N_CANDIDATES, body
    # The full regeneration pinned in the DB: one AI row per candidate.
    assert _folder_rows() == {
        "": (SUM_ROOT, False),
        FOLDER_A: (SUM_ALPHA, False),
        FOLDER_B: (SUM_BRavo, False),
    }

    # The re-fetch (the phase-77 re-show refresh — the nav re-click):
    # the synced catalog renders, and NO marker is anywhere — every
    # description-bearing row has a stored line.
    page.click("#nav-sources")
    _wait_top_level(page)
    src_row = page.locator("#folders-tbody tr")
    expect(src_row.locator("a.folder-link")).to_have_text(SOURCE)
    expect(src_row.locator("td:nth-child(2)")).to_have_text(str(N_FILES))
    expect(src_row.locator("td:nth-child(4) span")).to_have_text(SUM_ROOT)
    expect(page.locator(".kb-summary-pending")).to_have_count(0)

    # The gap: delete the bravo row AND the source-root row directly
    # (a historical fail-soft miss — the rows are gone from the app's
    # back, no KB change anywhere).
    _delete_folder_rows([(SOURCE, FOLDER_B), (SOURCE, "")])
    assert set(_folder_rows()) == {FOLDER_A}

    # The re-fetch: the two affected rows carry the D4 marker (the
    # pending set = exactly the missing rows — the D3 concept), the
    # intact folder shows its stored line and NO marker.
    page.click("#nav-sources")
    _wait_top_level(page)
    src_row = page.locator("#folders-tbody tr")
    s_span = src_row.locator("td:nth-child(4) span")
    expect(s_span).to_have_text(PENDING_COPY)
    # Phase 99 (task 01): the marker toggles onto the cell's base
    # .kb-desc-text span (the one-line clamp) — the class pair, exact.
    expect(s_span).to_have_class("kb-desc-text kb-summary-pending")
    expect(s_span).to_have_attribute("title", PENDING_TITLE)
    # The Edit button is KEPT with the marker (a manual save creates
    # the row and clears the marker in place).
    expect(src_row.locator(".kb-summary-edit")).to_be_visible()

    _drill(page, SOURCE)
    # The SOURCE level itself is pending too (its root row is gone) —
    # the level block shows the D4 pending note (title = the source
    # name — the source root is folder "").
    expect(page.locator("#kb-level")).to_be_visible()
    expect(page.locator("#kb-level-title")).to_have_text(SOURCE)
    expect(page.locator("#kb-level-summary")).to_have_text(PENDING_NOTE)
    rows = page.locator("#folders-tbody tr")
    expect(rows).to_have_count(2)
    alpha_row = rows.nth(0)
    bravo_row = rows.nth(1)
    expect(alpha_row.locator("a.folder-link")).to_have_text(FOLDER_A)
    # The intact folder: the stored line, never the marker.
    expect(alpha_row.locator("td:nth-child(4) span")).to_have_text(SUM_ALPHA)
    expect(alpha_row.locator("td:nth-child(4) span")).not_to_have_class(
        "kb-summary-pending"
    )
    # The affected folder: the marker (copy + class + title) — the
    # class pair with the phase-99 .kb-desc-text base span, exact.
    b_span = bravo_row.locator("td:nth-child(4) span")
    expect(b_span).to_have_text(PENDING_COPY)
    expect(b_span).to_have_class("kb-desc-text kb-summary-pending")
    expect(b_span).to_have_attribute("title", PENDING_TITLE)

    # Drill into the affected folder: ITS level block shows the D4
    # pending note (title = the full source-relative path).
    bravo_row.locator("a.folder-link").click()
    expect(page.locator("#kb-level-title")).to_have_text(f"{SOURCE}/{FOLDER_B}")
    expect(page.locator("#kb-level-summary")).to_have_text(PENDING_NOTE)

    # The row's Edit → Save a manual description → the marker is gone
    # IN PLACE (the cell shows the saved text — no reload, the D4
    # in-place clear).
    page.click(f'#kb-crumb a.kb-crumb-link:text-is("{SOURCE}")')  # up to the source level
    bravo_row = page.locator("#folders-tbody tr").nth(1)
    expect(bravo_row.locator("a.folder-link")).to_have_text(FOLDER_B)
    bravo_row.locator(".kb-summary-edit").click()
    page.fill(".kb-summary-editor", MANUAL_BRavo)
    page.click(".kb-summary-save")
    expect(bravo_row.locator(".kb-summary-status")).to_have_text(
        "Description updated."
    )
    b_span = bravo_row.locator("td:nth-child(4) span")
    expect(b_span).to_have_text(MANUAL_BRavo)
    expect(b_span).not_to_have_class("kb-summary-pending")

    # The second (UNCHANGED-KB) sync: the gap probe finds exactly the
    # source root missing (alpha stored, bravo manual) → the targeted
    # fill regenerates the root row only.
    body = _run_sync(fast_app_url)
    assert body["state"] == "success", body
    detail = body["detail"]
    assert detail["added"] == 0, detail
    assert detail["updated"] == 0 and detail["pruned"] == 0, detail
    assert detail["overview"] is False, detail  # unchanged → no overview burn
    # The terminal keeps the gap-fill's final counts (ONE missing row).
    assert body["summaries_done"] == 1, body
    assert body["summaries_total"] == 1, body
    # The DB pins the targeted fill: the root is BACK with its canned
    # text (AI-written again), bravo keeps the MANUAL row (untouched —
    # a regeneration would have rewritten the owner's words), alpha
    # untouched.
    rows_db = _folder_rows()
    assert rows_db[""] == (SUM_ROOT, False), rows_db
    assert rows_db[FOLDER_B] == (MANUAL_BRavo, True), rows_db
    assert rows_db[FOLDER_A] == (SUM_ALPHA, False), rows_db

    # The re-fetch: the marker is GONE — the current level (the source
    # root) shows the regenerated canned line in its level block, the
    # folder rows show the stored texts, and no marker anywhere.
    page.click("#nav-sources")
    expect(page.locator("#kb-level")).to_be_visible()
    expect(page.locator("#kb-level-title")).to_have_text(SOURCE)
    expect(page.locator("#kb-level-summary")).to_have_text(SUM_ROOT)
    rows = page.locator("#folders-tbody tr")
    expect(rows.nth(0).locator("td:nth-child(4) span")).to_have_text(SUM_ALPHA)
    expect(rows.nth(1).locator("td:nth-child(4) span")).to_have_text(MANUAL_BRavo)
    expect(page.locator(".kb-summary-pending")).to_have_count(0)
    # And at the top level: the source row's cell carries the
    # regenerated root line, no marker.
    page.locator("#kb-crumb a.kb-crumb-link").nth(0).click()
    src_row = page.locator("#folders-tbody tr")
    s_span = src_row.locator("td:nth-child(4) span")
    expect(s_span).to_have_text(SUM_ROOT)
    expect(s_span).not_to_have_class("kb-summary-pending")
    expect(page.locator(".kb-summary-pending")).to_have_count(0)