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

"""Phase 64 story E2E (Playwright): real-time progress for sync + upload.

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

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

The story gate for the phase's executable proof (owner-locked A1–A5):
both long-running KB jobs report **which file is being processed right
now** — not just "Syncing…"/"Uploading…" — and the archive upload is
fully **backgrounded**: ``POST /api/git-sources/upload`` answers 202 the
moment the archive is on disk (the "Successfully uploaded — <file>"
toast fires — the user may navigate away), the unpack/scan continues
server-side behind ``GET /api/git-sources/upload/status`` (the phase-32
``SyncStatus`` pattern), and the RAG-page sync button
(``/sources.html``) animates with the upload's current file while that
scan runs.

**Timing fixture (the phase's fixture note):** the mock LLM indexes
fast — the in-progress state is real but brief (a 25-file scan against
it takes ≈0.4 s, well under the UI's 2 s status poll). This module's
app therefore boots behind ``tests/e2e/slow_llm.py`` — a delay-injecting
reverse proxy in front of the mock LLM (``SLOW_DELAY_S`` per request →
a 25-file scan is 28 LLM requests ≈ 4.2 s), so the scan outlives the
2 s poll and the live-file label is asserted at BOTH layers the task
pins:

* **deterministic** — the status endpoints (``page.request`` / the
  concurrent recorder, ~100 ms cadence): ``state == "running"`` with a
  non-null ``current_file`` (``source/relative/path``) observed at some
  tick, the counts advancing, and the file-less ticks (unpack/row/
  model probe — A4) preceding the first file tick;
* **UI** — polling the button labels for the ``Importing`` /
  ``Syncing…`` / ``Processing…`` prefix plus a file path (generous
  timeout), which the pages' own 2 s poll ticks render.

The upload archive is built in-test with Python's ``tarfile`` from
**25 small ``.md`` files** (``e2e-prog.tar.gz`` → source ``e2e-prog``);
the sync subject is a host temp dir (``sync-corpus/``, 25 small
``.md`` files under ``notes/``) registered as a ``kind=local`` row —
the ``test_sync_button.py`` / ``test_local_directory_sources.py``
fixture styles. Per-module app env (the conftest pattern):
``BOR_UPLOAD_DIR`` scratch, ``BOR_GIT_SOURCES`` forced empty (the sync
sources are this suite's own local row), ``BOR_LLM_BASE_URL`` the slow
proxy.

Contract under test:

* **toast → navigate away (A2 + A3)**: on ``/git-sources.html`` the
  "Successfully uploaded — <source>" toast (``.toast.is-visible``,
  ``role="status"``) appears while the scan is still running;
  navigating to ``/sources.html`` shows the sync button animating
  (spinner + ``aria-busy``) with the ``Importing <file>`` label; on
  completion the button settles to "Sync sources" (no error UI,
  ``#sync-result`` stays empty — the upload's counts never render
  there, A3) and the catalog shows the uploaded documents (the
  phase-63 listing, untouched);
* **upload progress (A4)**: during the scan the status endpoint
  reports a non-null ``current_file`` (``source/relative/path`` shape,
  full denominator, advancing counts) at running ticks, the upload
  button shows "Processing… <file>" (bare "Processing…" during unpack)
  before the result line lands, and the toast fired earlier in the run
  — the result line itself comes from the status ``success``;
* **sync live file (A4)**: a multi-file local source; clicking
  **Sync sources** on ``/sources.html`` shows "Syncing…" (bare, the
  pre-64 click state) then "Syncing… <file> (n/m)" (both layers), then
  the pre-64 success settle — "Synced HH:MM" + the counts result line —
  preserved, plus the file in the label;
* **reload re-attach (A1 + A2)**: starting an upload and reloading
  ``/git-sources.html`` mid-scan leaves the button in the Processing
  state (disabled) with no error banner and NO second upload (the
  status endpoint's single run is still the one from before the
  reload — pinned on its ``started_at``); it then settles with the
  result line and the list shows exactly one row for the archive.

Test → story mapping (Playwright Mapping Rule):
1. ``test_upload_toast_then_navigate_away``
2. ``test_upload_progress_shows_current_file``
3. ``test_sync_live_file_label``
4. ``test_upload_reattach_after_reload``
"""
from __future__ import annotations

import io
import os
import re
import subprocess
import sys
import tarfile
import threading
import time
from collections.abc import Iterator
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.db import SessionLocal
from app.models import GitSource
from e2e.auth_helpers import login
from e2e.conftest import (
    ADMIN_PASSWORD,
    APP_PORT,
    MOCK_PORT,
    SESSION_SECRET,
    USE_REAL_LLM,
    _wait_http,
)

REPO = Path(__file__).resolve().parents[2]
APP_URL = f"http://127.0.0.1:{APP_PORT}"

GIT_SOURCES_URL = "/git-sources.html"
SOURCES_URL = "/sources.html"

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

#: Per-LLM-request delay on the proxy — the scan's duration becomes
#: deterministic: an N-file archive scan issues N + 3 LLM requests
#: (the check_models embed + chat probe, one embed per file, the
#: change-gated overview chat), so a 25-file upload takes ≈ 28 × 0.15 s
#: ≈ 4.2 s — long enough to outlive the UI's 2 s status poll (see the
#: module docstring's timing-fixture note).
SLOW_DELAY_S = "0.15"

#: The uploaded archive: 25 small docs under ``docs/`` (the phase's
#: fixture note — 20+ files so the scan outlives the 2 s poll).
UPLOAD_NAME = "e2e-prog"
UPLOAD_ARCHIVE = f"{UPLOAD_NAME}.tar.gz"
N_FILES = 25
UPLOAD_FILES: dict[str, str] = {
    f"docs/{i:02d}.md": f"# Doc {i:02d}\n\nDeterministic upload content {i:02d}.\n"
    for i in range(N_FILES)
}

#: The local sync source (host temp dir — the app runs on the same
#: machine): 25 small docs under ``notes/`` (the current_file's
#: source/relative/path shape has a directory level).
SYNC_SOURCE_DIR = "sync-corpus"
SYNC_FILES: dict[str, str] = {
    f"notes/{i:02d}.md": f"# Sync doc {i:02d}\n\nDeterministic sync content {i:02d}.\n"
    for i in range(N_FILES)
}

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

#: Generous settle budget: a 25-file scan against the slowed LLM is
#: ≈4.2 s; the UI's 2 s poll settles at most one tick after the
#: terminal state lands.
SETTLE_TIMEOUT_MS = 45_000


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


def _build_targz(path: Path, files: dict[str, str]) -> Path:
    """A deterministic ``.tar.gz`` (mtime 0) over the given files."""
    with tarfile.open(path, "w:gz") as tf:
        for rel, content in files.items():
            data = content.encode("utf-8")
            info = tarfile.TarInfo(rel)
            info.size = len(data)
            info.mtime = 0
            tf.addfile(info, io.BytesIO(data))
    return path


@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) — this suite's timing fixture: the live-file
    contract needs the scan to outlive 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 upload_dir(tmp_path_factory: pytest.TempPathFactory) -> Path:
    """The app's ``BOR_UPLOAD_DIR`` for this suite — a scratch dir the
    host-side assertions inspect (the app server runs on the same
    machine)."""
    return tmp_path_factory.mktemp("bor_uploads") / "uploads"


@pytest.fixture(scope="module")
def upload_archive(tmp_path_factory: pytest.TempPathFactory) -> Path:
    """The 25-file upload archive (``e2e-prog.tar.gz``)."""
    root = tmp_path_factory.mktemp("bor_archive")
    return _build_targz(root / UPLOAD_ARCHIVE, UPLOAD_FILES)


@pytest.fixture(scope="module")
def sync_local_dir(tmp_path_factory: pytest.TempPathFactory) -> Path:
    """The host temp dir the sync test registers as a ``kind=local``
    source — ``sync-corpus/notes/NN.md`` (25 files)."""
    root = tmp_path_factory.mktemp("bor_sync_local") / SYNC_SOURCE_DIR
    for rel, content in SYNC_FILES.items():
        path = root / rel
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text(content, encoding="utf-8")
    return root


@pytest.fixture(scope="module")
def app_server(
    mock_llm: int,
    slow_llm: int,
    upload_dir: Path,
    tmp_path_factory: pytest.TempPathFactory,
) -> Iterator[str]:
    """The real app under test — per-module env: the LLM base URL is the
    SLOW PROXY in front of the mock (the timing fixture), uploads unpack
    into a scratch dir, and the env git list is forced empty (the sync
    sources are this suite's own ``kind=local`` row, seeded per test)."""
    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 f"{SLOW_URL}/v1"
    )
    # Mock-calibrated threshold (conftest pattern) — no chat turn is
    # ever sent in this suite, but the app boots with the same env shape.
    env["BOR_RELEVANCE_THRESHOLD"] = "0.30"
    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
    env["BOR_GIT_SOURCES"] = ""
    env["BOR_UPLOAD_DIR"] = str(upload_dir)
    env["BOR_SOURCES_DIR"] = str(tmp_path_factory.mktemp("bor_checkouts"))
    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 per test (the E2E isolation pattern): the
    upload's/sync's counts and every ``/api/docs`` assertion must be
    this test's own doing. The E2E suites share one Postgres, and a
    leftover git_sources row or document would corrupt the row-count
    and doc-list assertions (and a leftover row would be a SECOND sync
    source, skewing the per-file counts)."""
    with SessionLocal() as db:
        db.execute(text("TRUNCATE chunks, documents, query_log, kb_overview, git_sources"))
        db.commit()


def _wait_no_running_jobs(app_url: str) -> None:
    """No background job may leak across tests (the run states live in
    the app's memory, and a still-running scan would keep importing
    into the NEXT test's truncated KB): wait for both status endpoints
    to be non-running BEFORE the truncate. Own admin session (the
    endpoints are admin-only) — usually a no-op: the tests settle only
    after their run's terminal state."""
    with httpx.Client(timeout=5.0) as client:
        client.post(f"{app_url}/api/login", json={"password": ADMIN_PASSWORD})
        for path in ("/api/sync/status", "/api/git-sources/upload/status"):
            body: dict[str, Any] = {}
            deadline = time.monotonic() + 90
            while time.monotonic() < deadline:
                r = client.get(f"{app_url}{path}")
                if r.status_code == 200:
                    body = r.json()
                    if body["state"] != "running":
                        break
                time.sleep(0.2)
            assert body["state"] != "running", (
                f"a background job was still running at test boundary: {path} {body}"
            )


@pytest.fixture(autouse=True)
def _clean(app_url: str, db_ready: None) -> Iterator[None]:
    _wait_no_running_jobs(app_url)
    _truncate_all()
    yield
    _truncate_all()


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


def _seed_local_source(path: str) -> None:
    """Register a ``kind=local`` source row directly (deterministic —
    the phase-49 page form is upload-only; a plain directory is an
    API/DB-only operation, the test_local_directory_sources.py note)."""
    with SessionLocal() as db:
        db.add(GitSource(url=path, kind="local", path=path))
        db.commit()


def _admin_git_sources_page(page: Page, app_url: str) -> None:
    """Real form login landing on the git sources page (admin settled:
    Sign out visible, the manager revealed by the page module)."""
    login(page, app_url, next=GIT_SOURCES_URL)
    expect(page).to_have_url(app_url + GIT_SOURCES_URL, timeout=30_000)
    expect(page.locator("#sign-out-btn")).to_be_visible(timeout=15_000)
    expect(page.locator("#git-sources-gate")).to_be_hidden()
    expect(page.locator("#git-sources-content")).to_be_visible()


def _status(page: Page, app_url: str, path: str) -> dict[str, Any]:
    """One (cookie-authenticated) status-endpoint fetch."""
    r = page.request.get(f"{app_url}{path}")
    assert r.status == 200, r.text
    return r.json()


def _wait_running_started_at(
    page: Page, app_url: str, path: str, timeout_s: float = 15.0
) -> str:
    """Poll the status endpoint until the run is ``running``; return its
    ``started_at`` (the run's identity — a second run would reset it)."""
    deadline = time.monotonic() + timeout_s
    body: dict[str, Any] = {}
    while time.monotonic() < deadline:
        body = _status(page, app_url, path)
        if body["state"] == "running":
            assert body["started_at"] is not None
            return str(body["started_at"])
        if body["state"] in ("success", "failed"):
            raise AssertionError(f"the run settled too fast to observe: {body}")
        time.sleep(0.1)
    raise AssertionError(f"the run never entered running: {body}")


class _TickRecorder:
    """The deterministic layer, concurrent with the UI assertions.

    A daemon thread that tight-polls (~100 ms cadence) the status
    endpoint with its OWN admin session (``httpx`` — the browser page
    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 (file-less
    unpack/row/probe phase, then the per-file 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."""

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

    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 browser context'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)
                            # The run status is in the app's memory and
                            # SURVIVES across this module's tests: a
                            # residual terminal state (a previous test's
                            # run) must not be mistaken for this test's
                            # own terminal — accept it only AFTER this
                            # run's "running" has been observed.
                            if body["state"] == "running":
                                saw_running = True
                            elif body["state"] in ("success", "failed") and 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 = 60.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 _assert_live_file_ticks(
    ticks: list[dict[str, Any]], source: str, n_files: int
) -> None:
    """A4 against the recorded running ticks (the deterministic layer):

    * the file-less ticks (unpack/row/model probe — before any file is
      indexed) come FIRST (the label is the bare prefix then);
    * SOME tick reports a non-null ``current_file`` in the
      ``source/relative/path`` shape;
    * ``files_total`` is the full pre-walk count from the first file
      tick, and ``files_done`` advances monotonically to it.
    """
    with_file = [t for t in ticks if t["current_file"]]
    assert with_file, f"no running tick carried a current_file: {ticks[:6]}"
    first_with = with_file[0]
    assert first_with["current_file"].startswith(f"{source}/"), first_with
    assert re.fullmatch(
        rf"{re.escape(source)}/.+\.(md|markdown)", first_with["current_file"]
    ), first_with
    assert first_with["files_total"] == n_files, first_with
    assert 1 <= first_with["files_done"] <= n_files, first_with
    dones = [t["files_done"] for t in with_file]
    assert dones == sorted(dones), f"files_done not monotonic: {dones}"
    # The last file tick is (n-1, n): the hook fires before the final
    # file's index, so the final count itself can land in the terminal
    # body instead of a recorded tick (100 ms cadence vs ≈160 ms file).
    assert max(dones) >= n_files - 1, f"files_done never advanced: {dones}"
    pre = [t for t in ticks if t["current_file"] is None]
    assert pre, f"no file-less running tick (the unpack phase): {ticks[:6]}"
    assert ticks.index(pre[0]) < ticks.index(first_with), (
        "a file tick preceded the file-less unpack ticks"
    )


# ---------------------------------------------------------------------------
# 1. Toast on 202 → navigate away → sync button animates with the
#    upload's current file → settle + catalog refresh (A2 + A3)
# ---------------------------------------------------------------------------


def test_upload_toast_then_navigate_away(
    page: Page, app_url: str, db_ready: None, upload_archive: Path
) -> None:
    """On ``/git-sources.html``: pick the multi-file archive, submit →
    the "Successfully uploaded — <source>" toast appears WHILE the scan
    is still running; immediately navigate to ``/sources.html`` → the
    sync button is present, animating (icon ``is-spinning``,
    ``aria-busy``) with the ``Importing`` label; wait for the settle →
    button idle ("Sync sources"), no error UI, and the catalog table
    shows the uploaded documents (the phase-63 listing, untouched)."""
    page.set_default_timeout(30_000)
    _admin_git_sources_page(page, app_url)
    # (A previous test's terminal run may re-render its result line at
    # boot — the task-05 re-attach contract; the submit below clears
    # it, and that is the "clean start" asserted after the click.)

    page.set_input_files("#archive-upload-file", str(upload_archive))
    page.click("#archive-upload-btn")

    # The 202 moment (A2): a single .toast node, visible, role=status,
    # naming the safe source name…
    toast = page.locator(".toast")
    expect(toast).to_have_count(1, timeout=SETTLE_TIMEOUT_MS)
    expect(toast).to_have_class(re.compile(r"\bis-visible\b"))
    assert toast.get_attribute("role") == "status"
    expect(toast).to_have_text(f"Successfully uploaded — {UPLOAD_NAME}")
    # …and the scan is still running — the result line is not up yet
    # (the toast precedes the scan's completion, A2).
    expect(page.locator("#archive-upload-result")).to_be_hidden()
    started_at = _wait_running_started_at(page, app_url, "/api/git-sources/upload/status")
    assert started_at is not None

    # Navigate away immediately (A1: the scan no longer dies with the
    # page).
    page.goto(app_url + SOURCES_URL)

    # The RAG-page sync button re-attaches to the in-flight upload scan
    # (A3): present, animating (spinner + aria-busy), disabled, with
    # the live "Importing <file>" label — no error UI on this page
    # (the upload's failure UI lives on the Sources page, A3).
    btn = page.locator("#sync-btn")
    expect(btn).to_be_visible(timeout=30_000)
    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(
        re.compile(rf"Importing {re.escape(UPLOAD_NAME)}/"), timeout=SETTLE_TIMEOUT_MS
    )
    expect(page.locator("#sync-error-banner")).to_be_hidden()

    # Settle: the button returns to idle, the sync-result line never
    # rendered the upload's counts (A3), and the catalog refreshes with
    # the uploaded documents — the phase-63 listing, untouched.
    expect(page.locator("#sync-label")).to_have_text("Sync sources", timeout=SETTLE_TIMEOUT_MS)
    expect(btn).to_be_enabled()
    expect(btn).not_to_have_attribute("aria-busy")
    expect(btn.locator(".sync-icon")).not_to_have_class(re.compile(r"\bis-spinning\b"))
    expect(page.locator("#sync-result")).to_have_text("")
    expect(page.locator("#docs-tbody tr")).to_have_count(N_FILES, timeout=30_000)
    expect(page.locator("#docs-tbody tr", has_text="docs/00.md")).to_have_count(1)
    expect(page.locator("#docs-tbody tr", has_text="docs/24.md")).to_have_count(1)
    expect(page.locator("#docs-tbody tr", has_text=UPLOAD_NAME)).to_have_count(N_FILES)


# ---------------------------------------------------------------------------
# 2. Upload progress: live current file at BOTH layers; the toast fired
#    earlier than the result (A4)
# ---------------------------------------------------------------------------


def test_upload_progress_shows_current_file(
    page: Page, app_url: str, db_ready: None, upload_archive: Path
) -> None:
    """During the scan: the status endpoint reports a non-null
    ``current_file`` (``source/relative/path`` shape) at some running
    tick; the upload button label shows "Processing…" with a file path
    (UI layer) BEFORE the result line lands; the toast fired earlier in
    the run (not after the result). The result line + the settle come
    from the status ``success``."""
    page.set_default_timeout(30_000)
    _admin_git_sources_page(page, app_url)

    recorder = _TickRecorder(app_url, "/api/git-sources/upload/status")
    recorder.start()
    page.set_input_files("#archive-upload-file", str(upload_archive))
    page.click("#archive-upload-btn")
    # A new attempt starts clean (the submit handler hides the result
    # line — any previous run's re-rendered line is gone by now).
    expect(page.locator("#archive-upload-result")).to_be_hidden()

    # The toast fires at the 202 — earlier in the run, NOT after the
    # result (the result line is still down when the toast is up).
    toast = page.locator(".toast")
    expect(toast).to_have_count(1, timeout=SETTLE_TIMEOUT_MS)
    expect(toast).to_have_class(re.compile(r"\bis-visible\b"))
    expect(toast).to_have_text(f"Successfully uploaded — {UPLOAD_NAME}")
    expect(page.locator("#archive-upload-result")).to_be_hidden()

    # UI layer: the button hands over to the scan — bare "Processing…"
    # at the 202 (A4: no file yet during the unpack phase)…
    btn = page.locator("#archive-upload-btn")
    expect(btn).to_be_disabled()
    expect(btn).to_have_text("Processing…", timeout=5_000)
    # …then the live file label ("Processing… <file> (n/m)") at the
    # page's next 2 s poll tick, still before the result line lands.
    expect(btn).to_have_text(
        re.compile(rf"Processing… {re.escape(UPLOAD_NAME)}/.+\.md \(\d+/{N_FILES}\)"),
        timeout=SETTLE_TIMEOUT_MS,
    )
    expect(page.locator("#archive-upload-result")).to_be_hidden()

    # Deterministic layer: the recorder's full tick series (the same
    # endpoint the UI's 2 s poll reads) — file-less unpack ticks first,
    # then the per-file ticks with the full denominator.
    terminal = recorder.stop()
    assert terminal["state"] == "success", terminal
    _assert_live_file_ticks(recorder.running_ticks, UPLOAD_NAME, N_FILES)
    # Phase 64: current_file is null in terminal states (the final
    # counts survive).
    assert terminal["current_file"] is None
    assert terminal["files_done"] == N_FILES
    assert terminal["files_total"] == N_FILES
    # The result line is rendered from the status success (the
    # UploadOut-shaped detail, counts unchanged in shape).
    assert terminal["detail"]["files"] == N_FILES
    assert terminal["detail"]["added"] == N_FILES
    assert terminal["detail"]["source"] == UPLOAD_NAME

    # Settle: the result line lands, the button restores, the input
    # cleared, and the list has exactly one row for the archive.
    result = page.locator("#archive-upload-result")
    expect(result).to_have_text(f"{N_FILES} added", timeout=30_000)
    expect(btn).to_be_enabled()
    expect(btn).to_have_text("Upload & scan")
    expect(page.locator("#archive-upload-file")).to_have_value("")
    expect(page.locator("#git-sources-tbody tr", has_text=UPLOAD_NAME)).to_have_count(1)


# ---------------------------------------------------------------------------
# 3. Sync live file label: both layers, then the preserved pre-64
#    success settle (A4)
# ---------------------------------------------------------------------------


def test_sync_live_file_label(
    page: Page, app_url: str, db_ready: None, sync_local_dir: Path
) -> None:
    """A multi-file local source (the ``test_sync_button.py`` fixture
    style): on ``/sources.html`` click **Sync sources** → the label
    shows "Syncing…" (bare, the pre-64 click state) then "Syncing…
    <file> (n/m)" while running (endpoint layer: ``current_file``
    non-null at running ticks; UI layer: the label poll), then the
    success settle with the counts result line — the pre-phase-64 sync
    UX is preserved, plus the file."""
    page.set_default_timeout(30_000)
    _seed_local_source(str(sync_local_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("Sync sources")

    recorder = _TickRecorder(app_url, "/api/sync/status")
    recorder.start()
    btn.click()

    # The click's immediate state is the pre-64 one (A4 — the bare
    # prefix until the import's first file): 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: the label gains the live file at the page's 2 s poll
    # tick ("Syncing… <source/relative/path> (n/m)").
    expect(page.locator("#sync-label")).to_have_text(
        re.compile(rf"Syncing… {re.escape(SYNC_SOURCE_DIR)}/.+\.md \(\d+/{N_FILES}\)"),
        timeout=SETTLE_TIMEOUT_MS,
    )

    # Deterministic layer: the recorder's full tick series — file-less
    # model-check ticks first, then the per-file ticks (full
    # denominator, advancing counts).
    terminal = recorder.stop()
    assert terminal["state"] == "success", terminal
    _assert_live_file_ticks(recorder.running_ticks, SYNC_SOURCE_DIR, N_FILES)
    assert terminal["current_file"] is None
    assert terminal["files_done"] == N_FILES
    assert terminal["files_total"] == N_FILES
    assert terminal["detail"]["added"] == N_FILES

    # The pre-64 success settle, preserved: "Synced HH:MM" + the counts
    # result line, button re-enabled, no error — and the catalog lists
    # the imported docs.
    expect(page.locator("#sync-label")).to_have_text(SYNCED_LABEL, timeout=30_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")
    expect(page.locator("#docs-tbody tr")).to_have_count(N_FILES, timeout=30_000)
    expect(page.locator("#docs-tbody tr", has_text="notes/00.md")).to_have_count(1)
    expect(page.locator("#docs-tbody tr", has_text=SYNC_SOURCE_DIR)).to_have_count(N_FILES)


# ---------------------------------------------------------------------------
# 4. Reload mid-scan → re-attach: no error, no second upload (A1 + A2)
# ---------------------------------------------------------------------------


def test_upload_reattach_after_reload(
    page: Page, app_url: str, db_ready: None, upload_archive: Path
) -> None:
    """Start the upload and, DURING the scan, reload
    ``/git-sources.html`` → the button is in the Processing state
    (disabled) with no error banner and NO second upload (the status
    endpoint's single run is still the one from before the reload —
    pinned on its ``started_at``); it then settles with the result line
    and the list shows exactly one row for the archive (in-place
    identity preserved)."""
    page.set_default_timeout(30_000)
    _admin_git_sources_page(page, app_url)

    page.set_input_files("#archive-upload-file", str(upload_archive))
    page.click("#archive-upload-btn")

    # The 202 toast (the run is in flight)…
    toast = page.locator(".toast")
    expect(toast).to_have_count(1, timeout=SETTLE_TIMEOUT_MS)
    expect(toast).to_have_class(re.compile(r"\bis-visible\b"))
    # …and the run's identity: the status's started_at (a second run
    # would reset it — the single-run claim is pinned on it).
    started_at = _wait_running_started_at(page, app_url, "/api/git-sources/upload/status")

    # Reload mid-scan — the page must re-attach, not dead-end.
    page.reload()
    expect(page).to_have_url(app_url + GIT_SOURCES_URL, timeout=30_000)
    expect(page.locator("#sign-out-btn")).to_be_visible(timeout=15_000)
    expect(page.locator("#git-sources-content")).to_be_visible()

    # The boot re-attach (task 05): the button is in the Processing
    # state (disabled, "Processing…") with no error banner and no
    # result line yet…
    btn = page.locator("#archive-upload-btn")
    expect(btn).to_be_disabled(timeout=15_000)
    expect(btn).to_have_text(re.compile(r"Processing…"), timeout=15_000)
    expect(page.locator("#archive-upload-error")).to_be_hidden()
    expect(page.locator("#archive-upload-result")).to_be_hidden()

    # …and it settles with the result line from the status success…
    expect(page.locator("#archive-upload-result")).to_have_text(
        f"{N_FILES} added", timeout=SETTLE_TIMEOUT_MS
    )
    expect(btn).to_be_enabled()
    expect(btn).to_have_text("Upload & scan")
    # …and the list shows exactly one row for the archive.
    expect(page.locator("#git-sources-tbody tr")).to_have_count(1)
    row = page.locator("#git-sources-tbody tr", has_text=UPLOAD_NAME)
    expect(row).to_have_count(1)
    expect(row.locator("span.git-source-kind")).to_have_text("Local")

    # No second upload: the terminal status is the SAME run — its
    # started_at is the one from before the reload.
    terminal = _status(page, app_url, "/api/git-sources/upload/status")
    assert terminal["state"] == "success", terminal
    assert str(terminal["started_at"]) == started_at, (
        f"the run's identity changed (a second upload ran): {terminal['started_at']}"
    )
    assert terminal["current_file"] is None