brain-of-reese/tests/e2e/test_sync_chunk_progress.py

"""Phase 129 task 05 E2E (Playwright, mock-only): while a file imports,
the sync status (endpoint + button label) shows the file's
embedding-chunk position — ``Syncing… chunkprog/big.md (1/2) · chunk
14/38`` — with ``chunks_total`` the file's TRUE content-chunk count and
``current_chunk`` the 1-based first chunk of the in-flight
token-budgeted batch (batch granularity, assumption A6), climbing 1 →
N through the file's embed batches, restarting per file, and cleared
(0/0, chunk part hidden) at every point the import phase ends.

The dedicated story suite for ``129_sync_chunk_progress`` (A16 — one
Playwright file per phase, run in isolation): the owner's report was
that ``Syncing… name/of/file/here (2/3)`` "needs to show what chunk
we're on out of how many".

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

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

MOCK-ONLY suite: ``E2E_REAL_LLM=1`` is not supported — the gate is the
deterministic mock plus the phase-64 ``tests/e2e/slow_llm.py`` delay
proxy that stretches the run past the recorder's ~100 ms cadence and
the UI's 2 s status poll, so every batch boundary and the rendered
labels are observed deterministically, not as races.

Timing fixture (the phase-98 sizing rationale, re-sized for batches):
``SLOW_DELAY_S = 1.5`` — the proxy delays EVERY LLM call, embeds
included, so each in-flight embed batch's position holds for ~1.5 s:
that outlives the tight-poll recorder's ~100 ms cadence by ~15× (every
batch boundary is densely sampled — the whole import span is sampled,
not just its ends), and the big file's import span (~30 batches
≈ 45 s) outlives the UI's 2 s poll ~20× (any window ≥ the poll period
catches a tick — the rendered-label assertions are deterministic, not
races). The whole changed sync is ≈ 35 LLM calls (2 model probes +
30 big-file embed batches + 1 small-file embed + 1 overview +
1 folder summary) ≈ 55–60 s — inside the generous settle budgets.

Sizing (why the file is ~60 KB): ``big.md`` is ~60 KB of markdown
(20 ``##`` sections × 3 ~1 KB paragraphs) → 61 content chunks at the
app's (code-default, pinned) 1200-char effective target — well over
the 30-chunk bar — which the token-budgeted batcher groups into ~30
requests of 2–3 chunks each: many enough batches that the climb is a
real sequence, few enough that the run stays inside the budgets. The
expected chunk count is computed IN THE TEST by calling
``chunk_document`` — the SAME function (and arguments: the importer's
``max(400, target)`` floor + the overlap setting) the importer's
``_replace_content_chunks`` uses — on the fixture file read back from
disk, and the expected batch boundaries by ``LLMClient._batch_texts``
on that list (the task-01 unit pattern). The three sizing knobs are
pinned to the code defaults in the app-under-test env (the leak-guard
pattern — an operator's gitignored ``.env`` tuning
``BOR_CHUNK_TARGET_CHARS`` must not change the suite's ground truth).

The suite runs ONE module app (the phase-79 full-inventory pattern —
it binds its own port) over the shared E2E Postgres: the SLOW app
(``BOR_LLM_BASE_URL`` = the slow proxy in front of the mock).

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): ``chunkprog/`` with EXACTLY TWO files, both
root-level (no subfolders → the summaries candidate set is just the
source root): ``big.md`` (the ~60 KB file above) + ``small.md`` (one
tiny chunk). The sorted walk processes ``big.md`` FIRST, so the file
counter reads ``(1/2)`` during the big span and ``(2/2)`` during the
small one — and the small file's single chunk is the proof that the
chunk position RESTARTS per file (the big file ends at its last
batch's first index, the small file reports ``(1, 1)``).

Autouse cleanup (the phase-96 pattern): before each test wait for no
running sync on the 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_endpoint_reports_the_chunk_position`` (endpoint — the slow
   leg, the phase-64 tight-poll ~100 ms recorder on
   ``GET /api/sync/status``): the prelude ticks are phase-less with
   the chunk fields 0/0; while ``phase == "import"`` and
   ``current_file == "chunkprog/big.md"`` — ``files_done/total ==
   1/2`` — every tick after the first batch has
   ``chunks_total == <the deterministic chunk_document count>`` and
   the deduped ``current_chunk`` sequence climbs STRICTLY from 1,
   every value one of the expected batch first-indices (batch
   granularity — assert the climb + the total, not a contiguous run),
   and the LAST tick before the file ends carries the final batch's
   first index (which equals ``chunks_total`` only when the last
   batch is a single chunk); while the tiny second file is in flight
   (``files_done/total == 2/2``) the position RESTARTS — a later tick
   has ``current_chunk == 1`` with ``chunks_total == 1``; the
   overview + summaries phase ticks are already cleared (0/0 — the
   import phase ends at the transition); the terminal has
   ``current_chunk == 0`` and ``chunks_total == 0`` while
   ``files_done == files_total == 2`` (the keep-final-file-counts
   split — the file counts keep their final values, the sub-file
   chunk position clears with the file, phase 129).
2. ``test_label_shows_the_chunk_part`` (UI — the slow leg, the page's
   own 2 s poll): after clicking **Sync sources**, the page's own
   poll renders the import span's ``#sync-label`` parts —
   ``.sync-label-main`` == ``Syncing… chunkprog/big.md``,
   ``.sync-label-count`` == ``(1/2)``, and ``.sync-label-chunk``
   VISIBLE matching ``· chunk \\d+/\\d+`` with the total == the
   expected chunk count — and the button ``title`` (and the
   ``#sync-progress`` box — the aria-live announcer, phase 150: the
   box is the single sync text surface) carries the UNTRUNCATED
   concatenation main + " " + count + " " + chunk (A4 — asserted
   atomically in the same in-browser evaluation, so the parts + title
   can never straddle a poll tick); after the file's embed span the
   CHUNK PART IS CLEARED — once the label has left the import phase
   (overview / summaries / the terminal), ``.sync-label-chunk`` is
   hidden + empty while the label itself keeps rendering in sync
   (the title tracks the rendered parts, A4); on settle the button
   reads the terminal ``Synced HH:MM`` (the phase-32 contract) with
   the count + chunk parts hidden and the ``#sync-progress`` box
   carrying the counts line (phase 150).
"""
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 text

from app.config import Settings as _Settings
from app.db import SessionLocal
from app.rag.chunker import chunk_document
from app.rag.llm import LLMClient
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 app binds its own port
# instead (a same-port second uvicorn dies on bind and would drive the
# wrong server). Env-overridable.
APP_PORT = int(os.environ.get("E2E_APP_PORT_CHUNKPROG", "8165"))
APP_URL = f"http://127.0.0.1:{APP_PORT}"

#: The slow-LLM proxy's port (the conftest's mock LLM stays on
#: MOCK_PORT — the proxy fronts it).
SLOW_PORT = int(os.environ.get("E2E_SLOW_LLM_PORT_CHUNKPROG", "8906"))
SLOW_URL = f"http://127.0.0.1:{SLOW_PORT}"

#: Per-LLM-request delay on the proxy — sized so each in-flight EMBED
#: BATCH's position outlives the recorder's ~100 ms cadence by ~15×
#: (the phase-64/98 rationale): the proxy delays EVERY call — embeds
#: included — so the whole import span is densely sampled, and the
#: big file's span (~45 s) outlives the UI's 2 s poll ~20× (see the
#: module docstring's timing fixture paragraph).
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 = "chunkprog"

#: The two fixture files (sorted walk: ``big.md`` FIRST → the file
#: counter reads (1/2) on the big span, (2/2) on the small one).
BIG_FILE = f"{SOURCE}/big.md"
SMALL_FILE = f"{SOURCE}/small.md"
N_FILES = 2

#: The big file's shape — 20 ``##`` sections × 3 ~1 KB paragraphs
#: (≈ 60 KB → 61 content chunks, 30+ at the default target; see the
#: module docstring's sizing paragraph).
N_BIG_SECTIONS = 20
N_BIG_PARAS = 3

#: The tiny second file — one chunk (``chunks_total == 1`` proves the
#: chunk position restarts per file).
SMALL_MD = "# Small\n\nA tiny second fixture file that chunks as one.\n"

#: The three sizing knobs, pinned to the CODE DEFAULTS in both the
#: app-under-test env (the leak guards below) and the test process's
#: expected-value computation — an operator's gitignored ``.env``
#: (which may legitimately tune ``BOR_CHUNK_TARGET_CHARS``) cannot
#: change the suite's ground truth. Derived from the class fields, so
#: they never drift from app/config.py (the conftest leak-guard
#: pattern).
CHUNK_TARGET = _Settings.model_fields["chunk_target_chars"].default
CHUNK_OVERLAP = _Settings.model_fields["chunk_overlap_chars"].default
EMBED_BATCH_SIZE = _Settings.model_fields["embed_batch_size"].default

#: The test process's mirror of the app-under-test sizing settings
#: (explicit kwargs beat any ``.env`` for the fields the expected
#: values depend on).
_EXPECTED_SETTINGS = _Settings(
    chunk_target_chars=CHUNK_TARGET,
    chunk_overlap_chars=CHUNK_OVERLAP,
    embed_batch_size=EMBED_BATCH_SIZE,
)

#: "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}$")
#: The settled boot label — 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).
IDLE_OR_SETTLED = re.compile(r"^(Sync sources|Synced \d{1,2}:\d{2})$")
#: The post-import label mains (overview / summaries / terminal) —
#: the states AFTER the big file's embed span where the chunk part
#: must be cleared while the label keeps rendering (test 2's step 2).
POST_IMPORT_MAIN_RE = re.compile(
    r"^(Writing KB overview…|Summarizing folders…(?: \S+)?|Synced \d{1,2}:\d{2})$"
)
#: The import-span chunk part (phase 129 task 04's unit-pinned shape):
#: "· chunk n/m" — ``m`` is asserted against the expected chunk count
#: by the in-browser snapshot (``n`` is batch-granular — any of the
#: expected first-indices).
CHUNK_PART_RE = re.compile(r"^· chunk (\d+)/(\d+)$")

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


def _big_md_content() -> str:
    """The deterministic ~60 KB big fixture file (see the module
    docstring's sizing paragraph): an H1 preamble (one chunk) + 20
    ``##`` sections of 3 ~1 KB single-line paragraphs each (the
    section's paragraph packing sub-splits each into ~3 chunks — no
    randomness, so the expected count is stable run to run)."""
    parts = ["# Big Fixture\n"]
    parts.append(
        "Intro paragraph for the big chunkprog fixture file. This preamble "
        "stays under the section target so it chunks as one.\n"
    )
    for k in range(1, N_BIG_SECTIONS + 1):
        parts.append(f"\n## Section {k}\n")
        for p in range(1, N_BIG_PARAS + 1):
            base = (
                f"Paragraph {p} of section {k} in the chunkprog big fixture "
                f"carries deterministic filler number {k * 100 + p} of the "
                f"large markdown file, written out to size. "
            )
            parts.append((base * 8)[:1050] + "\n\n")
    return "".join(parts)


def _expected_big_chunks(content: str) -> list[str]:
    """The big file's expected content chunks — the SAME function and
    arguments the importer's ``_replace_content_chunks`` uses
    (``chunk_document(content, rel, max(400, target), overlap)`` — the
    importer's token-cap floor rule; ``rel`` = the source-relative
    path): the endpoint's ``chunks_total`` must equal
    ``len(chunk_document(...))`` — the file's TRUE chunk count, not a
    batch count (the phase-129 A6 assumption)."""
    settings = _EXPECTED_SETTINGS
    target = max(400, settings.chunk_target_chars)
    return chunk_document(content, "big.md", target, settings.chunk_overlap_chars)


def _expected_batch_firsts(chunks: list[str]) -> list[int]:
    """The expected ``current_chunk`` values — the 1-based FIRST chunk
    of each token-budgeted embed batch, computed with the same
    ``LLMClient._batch_texts`` partition the app's embed loop walks
    (the task-01 unit pattern — call it directly for the boundaries)."""
    client = LLMClient(_EXPECTED_SETTINGS)
    firsts: list[int] = []
    acc = 0
    for batch in client._batch_texts(chunks):
        firsts.append(acc + 1)
        acc += len(batch)
    return firsts


def _dedup_consecutive(values: list[int]) -> list[int]:
    """Collapse consecutive duplicates (the ~100 ms recorder samples
    each 1.5 s batch window ~15× — the CLIMB lives in the distinct
    sequence)."""
    out: list[int] = []
    for v in values:
        if not out or out[-1] != v:
            out.append(v)
    return out


# ---------------------------------------------------------------------------
# The in-browser snapshot functions (Playwright Mapping Rule — the
# rendered DOM is the observable).
#
# Each returns null while the awaited label state is absent, and the
# PART texts + the button title once it is present — read in the SAME
# in-browser evaluation that verified the label state (a 2 s poll tick
# updates the parts + the title together in one JS task, so the atomic
# check can never straddle one — the phase-128 pattern).
# ---------------------------------------------------------------------------

#: Test 2, step 1 — the in-flight IMPORT label with the VISIBLE chunk
#: part: the main part carries "Syncing… chunkprog/big.md" (the
#: untruncated A4 text — the big file is the first of the two, so the
#: count part is "(1/2)"), the pinned chunk part is visible matching
#: "· chunk n/m" with m == the expected chunk count (the arg), and the
#: button title + the #sync-progress box (the aria-live announcer,
#: phase 150) carry the UNTRUNCATED concatenation
#: (main + " " + count + " " + chunk — A4: the announcer hears the
#: whole "Syncing… <file> (1/2) · chunk n/m").
_JS_IMPORT_CHUNK_SNAPSHOT = """
(n) => {
    const label = document.querySelector('#sync-label');
    const btn = document.querySelector('#sync-btn');
    if (!label || !btn) return null;
    const main = label.querySelector('.sync-label-main');
    const count = label.querySelector('.sync-label-count');
    const chunk = label.querySelector('.sync-label-chunk');
    if (!main || !count || !chunk || count.hidden || chunk.hidden) return null;
    if (main.textContent !== 'Syncing… chunkprog/big.md') return null;
    if (count.textContent !== '(1/2)') return null;
    const m = chunk.textContent.match(/^· chunk (\\d+)\\/(\\d+)$/);
    if (!m || Number(m[2]) !== n) return null;
    const full = main.textContent + ' ' + count.textContent + ' ' + chunk.textContent;
    if (btn.getAttribute('title') !== full) return null;
    const boxText = document.querySelector('#sync-progress-text');
    if (!boxText || boxText.textContent !== full) return null;
    return {
        mainText: main.textContent,
        countText: count.textContent,
        chunkText: chunk.textContent,
        title: btn.getAttribute('title'),
        boxText: boxText.textContent,
    };
}
"""

#: Test 2, step 2 — a state AFTER the big file's embed span (the
#: overview / summaries / terminal label — the import phase is over):
#: the CHUNK PART IS CLEARED — hidden + empty — while the label
#: itself keeps rendering in sync (the title carries the rendered
#: parts' concatenation, A4; the count part is present only in the
#: summaries label).
_JS_POST_IMPORT_SNAPSHOT = """
() => {
    const label = document.querySelector('#sync-label');
    const btn = document.querySelector('#sync-btn');
    if (!label || !btn) return null;
    const main = label.querySelector('.sync-label-main');
    const count = label.querySelector('.sync-label-count');
    const chunk = label.querySelector('.sync-label-chunk');
    if (!main || !count || !chunk) return null;
    const t = main.textContent;
    const postRe = /^(Writing KB overview…|Summarizing folders…(?: \\S+)?|Synced \\d{1,2}:\\d{2})$/;
    if (!postRe.test(t)) return null;
    if (!chunk.hidden || chunk.textContent !== '') return null;
    const countText = count.hidden ? '' : count.textContent;
    const full = t + (countText ? ' ' + countText : '');
    if (btn.getAttribute('title') !== full) return null;
    return {
        mainText: t,
        countText,
        chunkHidden: chunk.hidden,
        chunkText: chunk.textContent,
        title: btn.getAttribute('title'),
    };
}
"""


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


@pytest.fixture(scope="module")
def chunkprog_dir(tmp_path_factory: pytest.TempPathFactory) -> Path:
    """The one-source temp tree (see the module docstring): the app
    server runs on the same host, so the paths are visible to it.
    The directory NAME is the source name (``kind=local``, phase 38).
    Exactly two files: the ~60 KB ``big.md`` (30+ chunks) and the
    tiny one-chunk ``small.md`` (the per-file restart proof)."""
    root = tmp_path_factory.mktemp("bor_chunkprog")
    src = root / SOURCE
    src.mkdir()
    (src / "big.md").write_text(_big_md_content(), encoding="utf-8")
    (src / "small.md").write_text(SMALL_MD, 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 (the slow proxy in front of the mock),
    ``BOR_GIT_SOURCES`` forced empty (the sync source is this suite's
    own DB-registered local row), the sizing knobs pinned to the code
    defaults (the leak guards — see the module constants), and the
    other 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 — including the chunking / batching knobs this suite's
    # ground truth is computed from (see the module constants).
    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_CHUNK_TARGET_CHARS"] = str(CHUNK_TARGET)
    env["BOR_CHUNK_OVERLAP_CHARS"] = str(CHUNK_OVERLAP)
    env["BOR_EMBED_BATCH_SIZE"] = str(EMBED_BATCH_SIZE)
    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 the whole
    suite: it delays EVERY LLM call, embeds included, so each in-
    flight embed batch's position outlives the recorder's ~100 ms
    cadence by ~15× and the big file's import span outlives the UI's
    2 s poll ~20× (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,
    chunkprog_dir: Path,
    tmp_path_factory: pytest.TempPathFactory,
) -> Iterator[str]:
    """The SLOW app under test (the whole suite):
    ``BOR_LLM_BASE_URL`` is the slow proxy in front of the mock (the
    timing fixture — see the module docstring). ``chunkprog_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(APP_PORT), "--log-level", "warning"],
        cwd=REPO,
        env=env,
    )
    try:
        _wait_http(f"{APP_URL}/api/health")
        yield 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


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 file counts the syncs and the chunk-position
    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, 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, 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 (the module app is shared) from racing the truncate."""
    _wait_sync_settled(app_url)
    _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


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 app is 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×)."""

    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 = 180.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 _start_sync(page: Page, app_url: str, sync_dir: Path) -> None:
    """Register the source, land the admin on the sources view, wait
    for the settled boot label (the phase-64 pattern — also proves the
    phase-32 boot re-attach completed, so no later re-render can
    clobber the running state), then click **Sync sources** and pin
    the click's immediate 202 state (the bare "Syncing…" — the count
    AND the chunk parts hidden: no file yet, no batch in flight)."""
    _register_source(app_url, str(sync_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)
    expect(page.locator("#sync-label")).to_have_text(IDLE_OR_SETTLED)
    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, and
    # the label's MAIN part alone ("Syncing…" — the 202 click carries
    # no status, so both pinned parts hide, phase 128/129 writers).
    # Phase 149 (task 01): the running detail rides the box — the 202's
    # bare label is in #sync-progress (the button is the spinner).
    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-progress")).to_have_text("Syncing…")
    expect(page.locator(".sync-label-count")).to_be_hidden()
    expect(page.locator(".sync-label-chunk")).to_be_hidden()
    expect(page.locator("#sync-error-banner")).to_be_hidden()


def _wait_terminal(app_url: str, timeout_s: float = 180.0) -> dict[str, Any]:
    """Poll the status endpoint (own admin session) until the UI-
    triggered run reaches a terminal state — the source of truth for
    the settle (the UI's 2 s poll renders it at most one tick later).
    A FAILED terminal fails the test with the terminal body (the
    ``detail`` / error text) — never an opaque label timeout."""
    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
        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"):
                assert body["state"] == "success", body
                return body
            time.sleep(0.2)
    raise AssertionError(f"the run never reached a terminal state: {body}")


# ---------------------------------------------------------------------------
# 1. Endpoint — the chunk position climbs 1 → N through the big file's
#    embed batches (chunks_total = the file's true chunk count),
#    restarts per file, and clears at every point the import ends
# ---------------------------------------------------------------------------


def test_endpoint_reports_the_chunk_position(
    app_url: str, chunkprog_dir: Path, db_ready: None
) -> None:
    """A KB-changing sync of the two-file temp source is tight-polled
    by the ~100 ms recorder from the 202 to the terminal (the
    phase-64 pattern): the recorded ticks show the phase-129 chunk
    position end to end — the prelude is 0/0, the big file's import
    span reports ``chunks_total == <the deterministic chunk_document
    count>`` with ``current_chunk`` climbing strictly 1 → the final
    batch's first index (batch granularity — every recorded value is
    one of the expected ``_batch_texts`` boundaries, the sequence's
    LAST value before the file ends is the final batch's first index),
    the tiny second file RESTARTS the position (a later tick has
    ``current_chunk == 1`` with ``chunks_total == 1``), the
    overview/summaries ticks are already cleared (0/0 at the
    transitions), and the terminal keeps ``files_done == files_total
    == 2`` while the chunk position is 0/0 (the keep-final-file-
    counts split — the file counts survive, the sub-file position
    clears with the file)."""
    # The ground truth — computed from the fixture file ON DISK with
    # the SAME function + arguments the importer uses (see the helper
    # docstrings): the true chunk count + the batch first-indices.
    big_content = (chunkprog_dir / "big.md").read_text(encoding="utf-8")
    big_chunks = _expected_big_chunks(big_content)
    n_chunks = len(big_chunks)
    assert n_chunks >= 30, n_chunks  # the task's sizing bar (30+ chunks)
    firsts = _expected_batch_firsts(big_chunks)
    assert len(firsts) >= 3, firsts  # multi-batch — a real climb to assert

    _register_source(app_url, str(chunkprog_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=180)

    # --- the terminal: the keep-final-file-counts split — the FILE
    # counts keep their final values, the sub-file chunk position is
    # 0/0 (a sub-position of a file that's done — the phase-129
    # convention), and the file/phase clear like phase 64/98 ---
    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["current_chunk"] == 0, terminal
    assert terminal["chunks_total"] == 0, terminal
    assert terminal["phase"] is None, terminal

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

    # --- the prelude (model check): phase null, chunk fields 0/0
    # (reset with the run) — 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
    assert all(
        t["current_chunk"] == 0 and t["chunks_total"] == 0 for t in prelude
    ), prelude

    imp = [t for t in running if t["phase"] == "import"]
    assert imp, f"no import-phase tick: {[t['phase'] for t in running]}"
    big_idx = [i for i, t in enumerate(imp) if t["current_file"] == BIG_FILE]
    small_idx = [i for i, t in enumerate(imp) if t["current_file"] == SMALL_FILE]
    assert big_idx, f"no big-file import tick: {[t['current_file'] for t in imp][:8]}"
    assert small_idx, f"no small-file import tick: {[t['current_file'] for t in imp][:8]}"
    # The sorted walk processes big.md FIRST — and the files never
    # interleave: the big span ends before the small span starts.
    assert max(big_idx) < min(small_idx), (max(big_idx), min(small_idx))

    # --- the big file (files 1/2): chunks_total is the file's TRUE
    # chunk count (not a batch count) and current_chunk climbs
    # strictly through the batch boundaries ---
    big_ticks = [imp[i] for i in big_idx]
    assert all(
        t["files_done"] == 1 and t["files_total"] == N_FILES for t in big_ticks
    ), big_ticks[:4]
    with_total = [t for t in big_ticks if t["chunks_total"] > 0]
    assert with_total, f"no tick carried a chunk total: {big_ticks[:4]}"
    assert all(t["chunks_total"] == n_chunks for t in with_total), (
        [t["chunks_total"] for t in with_total], n_chunks
    )
    # The pre-first-batch ticks (if sampled) are 0/0 — no other
    # (chunk, total) combination is legal for this file.
    for t in big_ticks:
        if t["chunks_total"] == 0:
            assert t["current_chunk"] == 0, t
    seq = _dedup_consecutive([t["current_chunk"] for t in with_total])
    assert seq[0] == 1, f"the climb does not start at 1: {seq}"
    assert seq == sorted(set(seq)), f"the climb is not strictly increasing: {seq}"
    assert set(seq) <= set(firsts), f"off-partition positions: {seq} vs {firsts}"
    # The LAST tick before the file ends carries the final batch's
    # first index (== chunks_total only when the last batch is a
    # single chunk) — batch granularity, not a contiguous run.
    assert seq[-1] == firsts[-1], (seq, firsts)
    assert len(seq) >= 3, f"too few batch boundaries observed: {seq}"

    # --- the tiny second file (files 2/2): the position RESTARTS —
    # one chunk → the single batch reports (1, 1) ---
    small_ticks = [imp[i] for i in small_idx]
    assert all(
        t["files_done"] == N_FILES and t["files_total"] == N_FILES
        for t in small_ticks
    ), small_ticks[:4]
    with_total_s = [t for t in small_ticks if t["chunks_total"] > 0]
    assert with_total_s, f"no tick carried the small file's chunk total: {small_ticks[:4]}"
    assert all(
        t["current_chunk"] == 1 and t["chunks_total"] == 1 for t in with_total_s
    ), with_total_s[:4]

    # --- the phase transitions: the import phase ends at overview /
    # summaries — the chunk position is cleared (0/0) at the
    # transition (the two 1.5 s spans outlive the 100 ms cadence ~15×,
    # so the recorder samples both). ---
    ov = [t for t in running if t["phase"] == "overview"]
    sm = [t for t in running if t["phase"] == "summaries"]
    assert ov, f"no overview-phase tick: {[t['phase'] for t in running]}"
    assert sm, f"no summaries-phase tick: {[t['phase'] for t in running]}"
    for t in [*ov, *sm]:
        assert t["current_chunk"] == 0 and t["chunks_total"] == 0, t


# ---------------------------------------------------------------------------
# 2. UI — the label's chunk part: "Syncing… chunkprog/big.md (1/2) ·
#    chunk n/m" while a batch is in flight, the untruncated label on
#    the title + announcer, the part cleared after the file's span
# ---------------------------------------------------------------------------


def test_label_shows_the_chunk_part(
    page: Page, app_url: str, chunkprog_dir: Path, db_ready: None
) -> None:
    """A fresh admin page on ``/sources.html`` clicks **Sync sources**;
    the page's OWN 2 s poll renders the import span's ``#sync-label``
    as the phase-128/129 PARTS — the main part
    ``Syncing… chunkprog/big.md``, the pinned count part ``(1/2)``,
    and the VISIBLE chunk part matching ``· chunk n/m`` with the total
    == the expected chunk count (the big file's span is ~45 s under
    the slow proxy — the poll cannot miss it) — with the button
    ``title`` + the ``#sync-progress`` box (the aria-live announcer,
    phase 150) carrying the UNTRUNCATED concatenation (A4, asserted
    atomically in the same in-browser evaluation). After the file's
    embed span, once the label has left the import phase (overview /
    summaries / terminal), the chunk part is CLEARED — hidden + empty
    — while the label keeps rendering in sync (the title tracks the
    rendered parts); on settle the button reads the terminal
    ``Synced HH:MM`` (the phase-32 contract) with the count + chunk
    parts hidden and the ``#sync-progress`` box carrying the counts
    line (phase 150)."""
    page.set_default_timeout(30_000)
    big_content = (chunkprog_dir / "big.md").read_text(encoding="utf-8")
    n_chunks = len(_expected_big_chunks(big_content))
    assert n_chunks >= 30, n_chunks

    _start_sync(page, app_url, chunkprog_dir)

    # Step 1 — the import span with the chunk part (the page's own 2 s
    # poll; the big file's ~45 s span outlives the poll ~20×). The
    # in-browser snapshot verifies the parts + the title +
    # #sync-progress-text TOGETHER (one JS task per poll tick — the
    # atomic check can never straddle one) and returns the texts.
    snap = page.wait_for_function(
        _JS_IMPORT_CHUNK_SNAPSHOT, arg=n_chunks, timeout=SETTLE_TIMEOUT_MS
    ).json_value()
    assert snap["mainText"] == f"Syncing… {SOURCE}/big.md", snap
    assert snap["countText"] == "(1/2)", snap
    m = CHUNK_PART_RE.match(snap["chunkText"])
    assert m, snap
    assert int(m.group(2)) == n_chunks, snap  # the TRUE chunk count, not a batch count
    assert int(m.group(1)) >= 1, snap
    # A4: the untruncated concatenation rides the title AND the
    # aria-live announcer (the "· chunk" part included).
    full = f"Syncing… {SOURCE}/big.md (1/2) {snap['chunkText']}"
    assert snap["title"] == full and "· chunk" in snap["title"], snap
    # The box IS the announcer (phase 150) — it carries the same full
    # string the title does.
    assert snap["boxText"] == full, snap

    # Step 2 — after the file's embed span: the chunk part is CLEARED
    # (hidden + empty) once the label has left the import phase — the
    # overview / summaries / terminal states — while the label keeps
    # rendering (the title still tracks the rendered parts, A4). The
    # terminal is persistent, so the wait terminates deterministically
    # (the intermediate post-import states may be sampled or skipped
    # by the 2 s poll — any of them satisfies the pin).
    snap2 = page.wait_for_function(
        _JS_POST_IMPORT_SNAPSHOT, timeout=SETTLE_TIMEOUT_MS
    ).json_value()
    assert POST_IMPORT_MAIN_RE.match(snap2["mainText"]), snap2
    assert snap2["chunkHidden"] is True, snap2
    assert snap2["chunkText"] == "", snap2
    assert (
        snap2["title"]
        == snap2["mainText"] + (" " + snap2["countText"] if snap2["countText"] else "")
    ), snap2

    # The success settle (the phase-32 contract, preserved): "Synced
    # HH:MM" + the counts result line, the button re-enabled, the
    # count AND chunk parts hidden + empty, no error — the chunk part
    # rendered ONLY during the import phase (test 1 pins the
    # endpoint-side clearing at the transitions + terminal).
    _wait_terminal(app_url)
    expect(page.locator("#sync-label")).to_have_text(SYNCED_LABEL, timeout=30_000)
    btn = page.locator("#sync-btn")
    expect(btn).to_be_enabled()
    expect(btn).not_to_have_attribute("aria-busy")
    expect(page.locator(".sync-label-count")).to_be_hidden()
    expect(page.locator(".sync-label-count")).to_have_text("")
    expect(page.locator(".sync-label-chunk")).to_be_hidden()
    expect(page.locator(".sync-label-chunk")).to_have_text("")
    # The counts line rides the box (phase 150 — the sole sync surface).
    expect(page.locator("#sync-progress-text")).to_have_text(f"{N_FILES} added")
    expect(page.locator("#sync-error-banner")).to_be_hidden()