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

"""Phase 193 story E2E (Playwright): the sync label counts DOWN.

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

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

**Source (``TODO.md:7``):** "The sync status should show how many files
are left to sync, not how many files out of the total." The pre-phase
running label said ``Syncing… notes/a.md (1234/27000)`` — the owner's
position *in* the whole; the owner reads the distance *to the end*.
Phase 193 changes exactly ONE label part: ``fmtSyncLabel``'s count is
now ``N remaining`` with ``N = total − done``.

**Phase 194 (task 04, D5) retargets what the pair COUNTS — the label
builder stays byte-identical.** Phase 193 itself touched no server
code; phase 194 did, and this is the suite whose pins move with it.
``GET /api/sync/status`` no longer answers with the bare phase-64 key
set: it is that set plus the phase-187 memory pair plus the phase-194
scan pair (``scan_done`` / ``scan_total`` — the PRE-SCAN's own walk
position), and ``phase`` gained the ``"scan"`` stage between the null
prelude and ``"import"`` (the machine the page renders as
``Scanning… (n/total)``). More importantly the ``(done, total)`` pair
the label subtracts is no longer the WALK position: on a progress
caller (the UI sync is the only one) it is the position in the
importer's pre-scan PREDICTED change set — ``files_total`` the number
of files that will actually be re-indexed, ``files_done`` the 1-based
position in that set. ``N remaining`` therefore counts the WORK, which
is the fix for the owner's prod report of 2026-10-19 ("1216 remaining"
on a ~10,000-file KB with ~10 files to sync). The phase-193 wording —
the 1-based position across EVERY importable file of the stats
pre-walk — is exactly what ``BOR_SYNC_PENDING_COUNT=0`` restores, and
nothing below asserts it except as that rollback's history.

What survives the semantic shift UNEDITED is a property of this
suite's own fixture: every arm starts from a TRUNCATED KB, so the
prediction and the walk coincide (all ``N_FILES`` files are pending),
``files_total`` stays ``N_FILES`` and the ``1 … N−1`` count frames land
in the same slots phase 193 pinned — test 1's cross-layer invariant is
untouched arithmetic. The case where the two DIVERGE — a re-sync — is
test 4, and the phase's own suite
(``tests/e2e/test_sync_pending_count.py``, task 05) carries the K-of-N
incremental arm.

**A1 (owner-confirmed, phase 193):** ``N remaining`` shows while
``N ≥ 1`` and is HIDDEN at the final file (``N == 0``) — the main part
names that file; ``total === 0`` renders no count part either — and
since phase 194 there are TWO things that make it ``0``: the scan has
not reported a denominator yet (the null prelude / the ``0`` rollback,
test 2) and a run whose predicted change set is EMPTY (test 4).
**A2 (owner-confirmed):** the upload job shares the ONE builder, so it
shares the shape (``Importing <file> N remaining``) — and it gains
NOTHING from phase 194: the upload job runs no pre-scan and its
endpoint carries no scan fields (phase 90 A2), so test 3 is UNEDITED.

Fixture: a real git repo (the phase-28/32/107 ``file://`` transport-true
convention — ``BOR_GIT_SOURCES=file://<fixture repo>``, so the env
fallback IS the effective source under the autouse truncate) holding
``N_FILES = 5`` markdown docs — the first AND the last
multi-chunk (the phase-129 chunk-part climb rides a real embed span, and
the last one is the window whose label must carry NO count part), the
three in between one chunk each. ``tests/e2e/slow_llm.py`` delays every
LLM call ``SLOW_DELAY_S``, so a file's import span (content-embed
batches + the summary chat + the summary-chunk embed) is ≫ the UI's 2 s
status poll: every file gets its own rendered label, deterministically,
instead of racing it.

What the owner sees is captured, not sampled: a MutationObserver
installed before the click records EVERY value the single label writer
lands — the ``#sync-progress-text`` box (the phase-150 sole sync text
surface + sole ``aria-live`` region), the button ``title`` and
``aria-label`` (the phase-64 A4 untruncated placement) and the
``.sync-label-count`` span's visibility — so the assertions read the
whole rendered sequence, not the few instants a Python poll happens to
hit.

Arms (the phase's three, in test order):

1. ``test_running_label_counts_down_across_a_real_sync`` — **(a) the
   count-down**: every rendered import label matches
   ``^Syncing… <file> [1-N] remaining(?: · chunk n/m)?$``; the sequence
   of rendered ``N``s is strictly DECREASING (each new value smaller),
   never ``0``, never ``≥ total``; the ``· chunk n/m`` part coexists
   with it (its ``m`` equals the in-flight file's real
   ``chunk_document`` count — phase 129 unbroken); at the FINAL file no
   label carries a count part at all while the main part still names
   that file; no ``Syncing…`` label anywhere carries the pre-phase
   ``(n/m)`` tail; the ``(n/m)`` shape survives ONLY in the summaries
   phase (A1's exception) and, since phase 194, in the scan phase's own
   ``Scanning… (n/total)`` (never on a ``Syncing…`` label, and no scan
   label ever follows an import one); the overview label still paints;
   the terminal result line is byte-identical to the phase-64 counts
   line (the ``counts_line`` helper); and the endpoint's own tick
   series still reports the ``(done, total)`` pair whose subtraction
   equals every rendered ``N``. Phase 194 adds the machine's own
   evidence to that same tick series
   (``_assert_phase_machine_tick_series``): every body is exactly the
   pre-phase key set plus the memory pair plus the scan pair, the null
   prelude reports the scan pair as ``0/0``, a tick caught INSIDE the
   scan names no file and carries the full scan denominator, no scan
   tick follows a tick that already counted files, and the terminal
   KEEPS both pairs beside each other (the phase-64 keep-final-values
   convention) with ``scan_total == files_total == N_FILES`` on this
   fresh KB — the parity that lets the phase-193 frames stand. The scan
   window itself is five tiny files (single-digit milliseconds), so a
   100 ms tick is not guaranteed to land inside it: the deterministic
   ``Scanning… (n/total)`` frames are test 2's, and the choreographed
   server-side sequence lives in
   ``tests/integration/test_sync_scan_phase.py``.
2. ``test_prewalk_and_final_file_labels_have_no_count_part`` — **(c)
   the zero-total edges**, pinned on the exact payload the guards read:
   the page's own 2 s poll fed a ``phase: "scan"`` body renders
   ``Scanning… (n/total)`` (count span visible, title equal, never a
   file and never ``N remaining``), the same stage with
   ``scan_total: 0`` renders the bare ``Scanning…`` (the empty-KB
   guard), a ``phase: null`` / ``files_total: 0`` prelude renders the
   bare ``Syncing…`` (no count part, the count span hidden, the title
   equal), the first file renders ``N − 1 remaining``, a
   mid-run tick renders ``1 remaining`` *beside* ``· chunk 2/5``, the
   final-file tick (``done == total``) renders the file with the count
   part GONE and never brings it back, and the success terminal settles
   to the untouched counts line. The served bodies carry the status
   endpoint's key set — the phase-64 fields plus the memory and scan
   pairs, the stubs pinned against the same constant test 1 reads —
   because the client's own decision points (``total > 0``,
   ``total − done > 0``, ``scan_total > 0``) are what is under test.
3. ``test_upload_label_shares_the_count_down_builder`` — **(b) the
   upload arm (A2)**: the upload branch of the same two-job tree shares
   the builder, so a running upload status that carries counts renders
   ``Importing <file> N remaining`` and hides the part at the upload's
   final file, settling to the upload's own unchanged idle surface. The
   REAL background upload run is asserted first, against the live
   endpoint: since phase 90 it is unpack + register only, so it answers
   ``null``/``0``/``0`` for the whole run — which is exactly why the
   shared *shape* is pinned through a served status (the house
   ``page.route`` pattern) instead of a real upload, and why no real
   upload label can ever regress to ``(n/m)``. The upload branch never
   receives a status object (its endpoint has no phase / chunk / image
   fields — phase 129 / 152), so the chunk part coexists in the SYNC
   arms only (test 1 and 2); that is asserted here too. UNEDITED by
   phase 194 — the upload job runs no pre-scan and its endpoint gains
   no scan keys.
4. ``test_resync_of_an_unchanged_repo_reports_zero_work`` — **the
   phase-194 re-sync arm**, the run phase 193's fixture never had (the
   owner's prod case in its mildest form): sync this same repo TWICE
   and the second run predicts NOTHING, so its ``(done, total)`` pair
   is ``0/0`` on every tick — the A1 no-count-part exception driven by
   an EMPTY pending set instead of by a missing denominator — while the
   scan pair reports the whole KB walked (``scan_done == scan_total ==
   N_FILES``, kept at the terminal). That divergence (5 scanned, 0 to
   re-index) is the phase in one line, and it is exactly the shape that
   used to render "``N`` remaining" for files nobody would re-index.
   The terminal counts line stays the byte-identical
   ``0 added · N unchanged``. A no-change run makes no LLM call, so its
   import window is milliseconds and it renders no file label: the wire
   pair (``0/0``) plus test 2's ``total === 0`` guard pin is the honest
   form of "no count part" here.

Test → phase-task mapping (Playwright Mapping Rule, task 03 arms a/b/c
+ task 04's D5 retarget): 1 → (a) + the terminal-line pin + the
additive-payload + phase-machine tick-series pins,
2 → (c) + the A1 hidden-at-final-file edge + the ``Scanning…`` frames
(+ (a)'s arithmetic on exact inputs), 3 → (b), 4 → the D5 re-sync arm.
"""
from __future__ import annotations

import io
import json
import os
import re
import subprocess
import sys
import tarfile
import threading
import time
from collections.abc import Iterator
from datetime import UTC, 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
from app.db import SessionLocal
from app.rag.chunker import chunk_document
from e2e.auth_helpers import login
from e2e.conftest import (
    ADMIN_PASSWORD,
    MOCK_PORT,
    SESSION_SECRET,
    USE_REAL_LLM,
    _wait_http,
)
from e2e.sync_helpers import counts_line

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

#: This module's own port (the conftest session app owns 8123 in a
#: combined run — the phase-79 convention). Env-overridable.
APP_PORT = int(os.environ.get("E2E_APP_PORT_REMAINING", "8211"))
APP_URL = f"http://127.0.0.1:{APP_PORT}"
SLOW_PORT = int(os.environ.get("E2E_SLOW_LLM_PORT_REMAINING", "8914"))
SLOW_URL = f"http://127.0.0.1:{SLOW_PORT}"

SOURCES_URL = "/sources.html"

#: Per-LLM-request delay on the proxy (the phase-64/129 house value).
#: The run's PRELUDE (the two model probes) is ≈ 3 s and therefore
#: outlives the page's first 2 s tick (the bare-label contract the click
#: leaves up), and a one-chunk file costs 3 calls ≈ 4.5 s while a
#: multi-chunk file ≈ 10 s — every file gets several rendered labels of
#: its own (the module docstring's determinism note).
SLOW_DELAY_S = "1.5"

#: The module's wall-time budget (app boot + two full 5-file syncs — the
#: phase-194 re-sync arm runs the fixture's sync twice — plus the two
#: served-status arms). A breach is a speed canary, not flake tolerance
#: (the phase-64 convention).
WALL_BUDGET_S = 480.0

#: The sync's subject: a git repo whose DIRECTORY NAME is the source
#: name (the ``repo_name`` basename rule, phase 28). Deliberately NOT
#: named after the part under test — a source path containing the word
#: "remaining" would make every "does this label carry the count part?"
#: check match the path itself.
SOURCE = "countdown"
N_FILES = 5

#: The walk order is the sorted relative path, so the names below pin
#: the ORDER the owner watches: the first file (multi-chunk → the chunk
#: part climbs while the count reads ``4 remaining``), three single-chunk
#: files (3 → 2 → 1 remaining), the LAST file (multi-chunk → its own
#: chunk part, NO count part — the A1 hidden edge).
FIRST_REL = "notes/a-multi-first.md"
LAST_REL = "notes/z-multi-last.md"
REL_ORDER = [FIRST_REL, "notes/b-second.md", "notes/c-third.md", "notes/d-fourth.md", LAST_REL]

#: The count-bearing values the label may ever show (A1): ``1 … N−1``.
MAX_COUNT = N_FILES - 1

#: The archive the upload arm uploads (the phase-98 fixture convention —
#: the same kind of content as the sync fixture, packaged).
UPLOAD_NAME = "uploadkb"
UPLOAD_FILES: dict[str, str] = {
    f"docs/{i:02d}.md": f"# Upload doc {i:02d}\n\nDeterministic upload body {i:02d}.\n"
    for i in range(3)
}

#: The chunker settings the app under test is pinned to (the leak-guard
#: convention): the expected ``chunks_total`` is computed with the SAME
#: function and arguments the importer uses, so a local ``.env`` cannot
#: move the number the suite asserts.
CHUNK_TARGET = int(Settings.model_fields["chunk_target_chars"].default)
CHUNK_OVERLAP = int(Settings.model_fields["chunk_overlap_chars"].default)

#: The import-phase label grammar since phase 193 — the main part (the
#: prefix + the ``source/relative/path``), the OPTIONAL count part, the
#: OPTIONAL phase-129 chunk part. ``<file>`` is kept loose on purpose
#: where a suite-wide grammar is the point; the per-file assertions
#: below re-anchor it on this suite's own paths.
IMPORT_LABEL_RE = re.compile(
    r"^Syncing… \S+(?: [0-9]+ remaining)?(?: · chunk \d+/\d+)?$"
)
#: The scan-phase label grammar phase 194 (task 03, D4) adds: the bare
#: ``Scanning…`` (no denominator yet — the empty KB and the scan's own
#: first frame) or it plus the ``(n/total)`` count part. NEVER a file
#: (``current_file`` is null through the phase) and never ``N
#: remaining`` — the walk position is honest behind the word "Scanning",
#: never behind "Syncing" (TODO L6).
SCAN_LABEL_RE = re.compile(r"^Scanning…(?: \(\d+/\d+\))?$")
#: The pre-phase shape phase 193 removed from this part — no ``Syncing…``
#: label may ever carry a ``(n/m)`` tail again. Phase 194 leaves that
#: true: the ``(n/m)`` shape is legal ONLY on a ``Scanning…`` or a
#: ``Summarizing folders…`` label (and no label starting ``Syncing…``
#: matches either).
OLD_FRACTION_RE = re.compile(r"\(\d+/\d+\)")
SYNCED_LABEL = re.compile(r"^Synced \d{1,2}:\d{2}$")

#: The FULL ``GET /api/sync/status`` key set this suite's stubs mirror and
#: every real tick is pinned against — the phase-64/98/129/152 fields,
#: the phase-187 memory pair and the phase-194 SCAN pair (``scan_done`` /
#: ``scan_total``). The additive chain, pinned as ONE constant so a future
#: key has to be added here (and to the stub below) instead of drifting
#: away from the client's view of the payload.
EXPECTED_STATUS_KEYS = frozenset(
    {
        "state",
        "started_at",
        "finished_at",
        "current_file",
        "files_done",
        "files_total",
        "scan_done",
        "scan_total",
        "phase",
        "current_summary",
        "summaries_done",
        "summaries_total",
        "current_chunk",
        "chunks_total",
        "images_done",
        "images_total",
        "detail",
        "error",
        "rss_mb",
        "rss_peak_mb",
    }
)

SETTLE_TIMEOUT_MS = 180_000


def _now_iso() -> str:
    """A "just finished" stamp for the served terminals — the phase-154
    same-day rule renders it as the plain ``Synced HH:MM`` (computed per
    call, so a module that runs across midnight stays honest)."""
    return datetime.now(UTC).isoformat()


# ---------------------------------------------------------------------------
# The fixture KB
# ---------------------------------------------------------------------------


def _tiny_md(title: str) -> str:
    """A one-chunk document (its import span = 3 slowed LLM calls)."""
    return f"# {title}\n\nDeterministic one-chunk body for the {title}.\n"


def _multi_md(tag: str, sections: int = 4) -> str:
    """A multi-chunk document: an H1 preamble (one chunk) + *sections*
    ``##`` sections of ~1500 chars each (one chunk apiece at the
    code-default 2000-char target, and one embed batch apiece at the
    700-token request budget) — no randomness, so the chunk count is
    stable run to run (the phase-129 fixture idiom)."""
    parts = [f"# {tag} multi-chunk fixture document\n"]
    parts.append(
        f"Preamble paragraph of the {tag} remaining-count fixture. This "
        "paragraph sits under the H1 and chunks on its own.\n"
    )
    for k in range(1, sections + 1):
        parts.append(f"\n## Section {k}\n")
        base = (
            f"Section {k} paragraph of the {tag} remaining-count fixture "
            "carries deterministic filler for the chunker: alpha bravo "
            "charlie delta echo foxtrot golf hotel india juliet kilo lima "
            "mike november oscar papa quebec romeo sierra tango uniform "
            "victor whiskey xray yankee zulu number "
        )
        parts.append((base * 4)[:1500] + "\n\n")
    return "".join(parts)


def _fixture_files() -> dict[str, str]:
    return {
        FIRST_REL: _multi_md("first"),
        REL_ORDER[1]: _tiny_md("second"),
        REL_ORDER[2]: _tiny_md("third"),
        REL_ORDER[3]: _tiny_md("fourth"),
        LAST_REL: _multi_md("last"),
    }


def _expected_content_chunks(rel: str) -> int:
    """The file's TRUE content-chunk count — the SAME function and
    arguments the importer's chunk replacement uses (the phase-129
    convention: the endpoint's ``chunks_total`` is the chunk count, not
    a batch count)."""
    content = _fixture_files()[rel]
    return len(chunk_document(content, rel, max(400, CHUNK_TARGET), CHUNK_OVERLAP))


def _git(cwd: Path, *args: str) -> None:
    """Run git in *cwd*; a non-zero exit fails the fixture loudly."""
    proc = subprocess.run(["git", *args], cwd=cwd, capture_output=True, text=True)
    if proc.returncode != 0:
        raise AssertionError(f"git {' '.join(args)} failed: {proc.stderr.strip()}")


@pytest.fixture(scope="module")
def remaining_repo(tmp_path_factory: pytest.TempPathFactory) -> Path:
    """The real git repo the sync clones over ``file://`` (5 markdown
    docs — see ``REL_ORDER``). ``tmp_path_factory`` because the
    module-scoped app needs it for the module's lifetime."""
    root = tmp_path_factory.mktemp("bor_remaining")
    repo = root / SOURCE
    for rel, content in _fixture_files().items():
        target = repo / rel
        target.parent.mkdir(parents=True, exist_ok=True)
        target.write_text(content, encoding="utf-8")
    _git(repo, "init", "-q")
    _git(repo, "add", "-A")
    _git(
        repo,
        "-c", "user.email=e@x", "-c", "user.name=t", "-c", "commit.gpgsign=false",
        "commit", "-qm", "remaining-count fixture",
    )
    assert (repo / ".git").is_dir()
    assert len(list(repo.rglob("*.md"))) == N_FILES
    return repo


@pytest.fixture(scope="module")
def upload_archive(tmp_path_factory: pytest.TempPathFactory) -> Path:
    """The upload arm's archive (the ``test_sync_upload_progress.py``
    convention: the same shape of content as the sync fixture, packed as
    a deterministic ``.tar.gz``)."""
    root = tmp_path_factory.mktemp("bor_remaining_archive")
    path = root / f"{UPLOAD_NAME}.tar.gz"
    with tarfile.open(path, "w:gz") as tf:
        for rel, content in UPLOAD_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


def _app_env(
    base_url: str, repo: Path, upload_dir: Path, tmp_path_factory: pytest.TempPathFactory
) -> dict[str, str]:
    """The per-module app env (the conftest pattern): the slow proxy is
    the LLM, ``BOR_GIT_SOURCES`` is this suite's own ``file://`` repo (the
    env fallback IS the effective source — the autouse truncate keeps the
    ``git_sources`` table empty), and the chunker knobs are pinned to the
    code defaults so the expected ``chunks_total`` stays predictable
    (process env outranks an operator's gitignored ``.env``)."""
    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
    )
    env["BOR_RELEVANCE_THRESHOLD"] = "0.30"
    env["BOR_LEXICAL_SUPPORT_FLOOR"] = "0.15"
    env["BOR_SOURCE_USEFULNESS_FLOOR"] = "0.15"
    env["BOR_LLM_RETRY_DELAY"] = "0"
    env.setdefault(
        "BOR_DATABASE_URL",
        "postgresql+psycopg://reese:reese@localhost:5432/brain_of_reese",
    )
    env["BOR_ADMIN_PASSWORD"] = ADMIN_PASSWORD
    env["BOR_SESSION_SECRET"] = SESSION_SECRET
    env["BOR_GIT_SOURCES"] = f"file://{repo}"
    env["BOR_SOURCES_DIR"] = str(tmp_path_factory.mktemp("bor_checkouts"))
    env["BOR_UPLOAD_DIR"] = str(upload_dir)
    env["BOR_CHUNK_TARGET_CHARS"] = str(CHUNK_TARGET)
    env["BOR_CHUNK_OVERLAP_CHARS"] = str(CHUNK_OVERLAP)
    env["BOR_DOCS_REPO"] = ""
    env["BOR_SUGGESTIONS"] = json.dumps(["Ask something neutral"])
    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 suite's timing fixture: it delays
    EVERY LLM call, embeds included, so each file's import span outlives
    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,
    remaining_repo: Path,
    tmp_path_factory: pytest.TempPathFactory,
) -> Iterator[str]:
    """The app under test — the slow one, on its own port."""
    env = _app_env(
        f"{SLOW_URL}/v1",
        remaining_repo,
        tmp_path_factory.mktemp("bor_remaining_uploads") / "uploads",
        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 per test (the E2E isolation pattern): the
    suites share one Postgres, and a leftover document would move the
    sync's ``added`` counts the terminal line pins."""
    with SessionLocal() as db:
        db.execute(
            text(
                "TRUNCATE chunks, documents, query_log, steering_notes, "
                "kb_overview, git_sources, folder_summaries"
            )
        )
        db.commit()


def _wait_sync_settled(base_url: str, timeout_s: float = 240.0) -> None:
    """No run may leak across a test boundary (the run state lives in the
    app's memory and a running run keeps importing into the NEXT test's
    truncated KB). Own admin session; usually a no-op."""
    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
            body = r.json()
            if body["state"] == "failed":
                print(f"sync remaining-count: FAILED terminal at a boundary: {body.get('error')}")
            if body["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]:
    _wait_sync_settled(app_url)
    _truncate_all()
    yield
    _truncate_all()


@pytest.fixture(scope="module", autouse=True)
def _wall_time_guard() -> Iterator[None]:
    start = time.monotonic()
    yield
    elapsed = time.monotonic() - start
    assert elapsed < WALL_BUDGET_S, (
        f"the module took {elapsed:.1f}s — over the {WALL_BUDGET_S:.0f}s wall budget"
    )


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

#: The in-browser recorder: every value the ONE label writer lands, with
#: the surfaces phase 64's A4 pins together (the box text, the button
#: title, the button aria-label) plus the count span's own visibility —
#: read in one JS task, so a frame can never straddle two writes.
_INSTALL_RECORDER = """
() => {
  if (window.__borFrames) return window.__borFrames.length;
  window.__borFrames = [];
  const box = document.querySelector('#sync-progress');
  const textEl = document.querySelector('#sync-progress-text');
  const btn = document.querySelector('#sync-btn');
  const countEl = document.querySelector('.sync-label-count');
  const rec = () => {
    const frame = {
      text: textEl.textContent,
      title: btn.getAttribute('title') || '',
      aria: btn.getAttribute('aria-label') || '',
      count: countEl ? countEl.textContent : '',
      countHidden: countEl ? countEl.hidden : null,
      hidden: box.hidden,
    };
    const last = window.__borFrames[window.__borFrames.length - 1];
    if (last
        && last.text === frame.text
        && last.title === frame.title
        && last.aria === frame.aria
        && last.countHidden === frame.countHidden) {
      return;
    }
    window.__borFrames.push(frame);
  };
  const mo = new MutationObserver(rec);
  mo.observe(textEl, { childList: true, characterData: true, subtree: true });
  mo.observe(btn, { attributes: true, attributeFilter: ['title', 'aria-label'] });
  if (countEl) mo.observe(countEl, { attributes: true, attributeFilter: ['hidden'] });
  rec();
  return window.__borFrames.length;
}
"""


def _install_recorder(page: Page) -> None:
    page.evaluate(_INSTALL_RECORDER)


def _frames(page: Page) -> list[dict[str, Any]]:
    frames: list[dict[str, Any]] = page.evaluate("() => window.__borFrames")
    assert frames, "the in-browser recorder captured no label frames"
    return frames


def _texts(frames: list[dict[str, Any]]) -> list[str]:
    """The recorded label sequence, empty frames dropped (the box's
    idle-hide is not a label)."""
    return [f["text"] for f in frames if f["text"]]


class _TickRecorder:
    """The deterministic layer: a daemon thread tight-polling (~100 ms)
    the SAME status endpoint the UI's 2 s poll reads, with its OWN admin
    session (the Playwright sync API is not thread-safe, so the thread
    never touches the page). It starts nothing — it only watches.

    ``wait_for_running`` is the phase-194 re-sync arm's need: a recorder
    started while a PREVIOUS run's terminal is still on the endpoint must
    not treat that body as its own. A body belongs to THIS run once the
    recorder has seen it go ``running``, or — the belt-and-braces half,
    for a run faster than the 100 ms cadence — once its own
    ``started_at`` post-dates the recorder's start (the
    ``test_fast_path_deferred_content.py`` idiom). Without the flag every
    body counts, which is what the module's first arm wants (there is no
    previous run to confuse it with).
    """

    def __init__(self, base_url: str, path: str, *, wait_for_running: bool = False) -> None:
        self._url = f"{base_url}{path}"
        self._login_url = f"{base_url}/api/login"
        self._wait_for_running = wait_for_running
        self._saw_running = not wait_for_running
        self._started_after = 0.0
        self._ticks: list[dict[str, Any]] = []
        self._terminal: dict[str, Any] | None = None
        self._stop = threading.Event()
        self._thread: threading.Thread | None = None

    @staticmethod
    def _epoch(body: dict[str, Any]) -> float | None:
        """The body's ``started_at`` as epoch seconds (``None`` when no
        run ever started, or the stamp is unreadable)."""
        raw = body.get("started_at")
        if not raw:
            return None
        try:
            return datetime.fromisoformat(str(raw)).timestamp()
        except ValueError:  # pragma: no cover — defensive, an odd stamp
            return None

    def _is_ours(self, body: dict[str, Any]) -> bool:
        """Does this body belong to the run this recorder was started for?"""
        if body["state"] == "running":
            self._saw_running = True
        if self._saw_running:
            return True
        started = self._epoch(body)
        return started is not None and started >= self._started_after - 1.0

    def start(self) -> None:
        if self._wait_for_running:
            self._started_after = time.time()

        def run() -> None:
            with httpx.Client(timeout=5.0) as client:
                client.post(self._login_url, json={"password": ADMIN_PASSWORD})
                while not self._stop.is_set():
                    try:
                        r = client.get(self._url)
                        if r.status_code == 200:
                            body = r.json()
                            if not self._is_ours(body):
                                # A previous run's terminal — not ours.
                                self._stop.wait(0.02)
                                continue
                            self._ticks.append(body)
                            if body["state"] in ("success", "failed", "cancelled"):
                                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 = 240.0) -> dict[str, Any]:
        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, (
            f"the recorder saw no terminal state (last: {self._ticks[-3:]})"
        )
        return self._terminal

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

    @property
    def ticks(self) -> list[dict[str, Any]]:
        """EVERY body this recorder saw, the terminal included (the
        phase-194 phase-machine pin reads the whole series, not just the
        running frames)."""
        return list(self._ticks)


def _serve_statuses(page: Page, stages: dict[str, Any]) -> None:
    """Route the two job-status endpoints to the CURRENT stage body (the
    house ``page.route`` pattern). ``stages`` is mutated by the test
    between the UI waits, so each 2 s poll of the REAL page renders
    exactly the payload the test put in front of it — the client's own
    decision points, pinned without a timing race.

    ``stages`` keys: ``sync`` / ``upload`` → the body dicts. A ``None``
    value falls through to the real endpoint."""

    def handle(route: Any) -> None:
        path = route.request.url
        if "/api/sync/status" in path:
            body = stages.get("sync")
        elif "/api/git-sources/upload/status" in path:
            body = stages.get("upload")
        else:  # pragma: no cover — only the two patterns are routed
            route.fallback()
            return
        if body is None:
            route.fallback()
            return
        route.fulfill(status=200, content_type="application/json", body=json.dumps(body))

    page.route("**/api/sync/status", handle)
    page.route("**/api/git-sources/upload/status", handle)


def _sync_status(**overrides: Any) -> dict[str, Any]:
    """One ``GET /api/sync/status`` body with the endpoint's CURRENT key
    set (the phase-64/98/129/152/187 fields the client reads plus the
    phase-194 scan pair — the additive chain ``EXPECTED_STATUS_KEYS``
    pins) — the served stubs are faithful to the payload shape; only the
    values are the test's. The equality against that constant is asserted
    below so a stub can never drift from the real endpoint unnoticed."""
    body: dict[str, Any] = {
        "state": "idle",
        "started_at": None,
        "finished_at": None,
        "current_file": None,
        "files_done": 0,
        "files_total": 0,
        # Phase 194 (task 02): the pre-scan's position, additive after
        # files_total and 0/0 until the scan reports (the degenerate and
        # BOR_SYNC_PENDING_COUNT=0 edges both render no count part).
        "scan_done": 0,
        "scan_total": 0,
        "phase": None,
        "current_summary": None,
        "summaries_done": 0,
        "summaries_total": 0,
        "current_chunk": 0,
        "chunks_total": 0,
        "images_done": 0,
        "images_total": 0,
        "detail": {},
        "error": None,
        "rss_mb": None,
        "rss_peak_mb": None,
    }
    assert set(body) == EXPECTED_STATUS_KEYS, set(body) ^ EXPECTED_STATUS_KEYS
    body.update(overrides)
    return body


def _upload_status(**overrides: Any) -> dict[str, Any]:
    body: dict[str, Any] = {
        "state": "idle",
        "started_at": None,
        "finished_at": None,
        "current_file": None,
        "files_done": 0,
        "files_total": 0,
        "detail": {},
        "error": None,
    }
    body.update(overrides)
    return body


def _assert_phase_machine_tick_series(
    ticks: list[dict[str, Any]],
    terminal: dict[str, Any],
    *,
    scan_total: int,
    files_total: int,
    prelude_zero: bool = True,
) -> None:
    """The phase-194 machine, read off the status endpoint the UI polls.

    Every claim here is deterministic on a real run, which is why none of
    them depends on a tick happening to land inside the scan (a five-file
    scan is milliseconds — see the module docstring):

    * the payload is ADDITIVE — exactly :data:`EXPECTED_STATUS_KEYS` on
      every body, idle/running/terminal alike;
    * the null prelude reports the scan pair as ``0/0`` (the run has not
      reached ``import_sources`` yet — the scan is what sets it);
    * a tick caught inside ``phase == "scan"`` names NO file, carries the
      full ``scan_total`` denominator and a position that only climbs;
    * the machine is ONE-DIRECTIONAL — no scan tick follows a tick that
      already counted files (``files_total > 0`` means the first
      ``progress`` call flipped it, and the phase never flips back);
    * the terminal CLEARS ``phase`` and KEEPS both pairs side by side
      (the phase-64 keep-final-values convention);
    * ``files_total`` is never larger than the scan's denominator on a
      run whose KB did not change mid-run — the count and its denominator
      stay on the SAME predicted set.
    """
    for tick in ticks:
        assert set(tick) == EXPECTED_STATUS_KEYS, (
            f"the status payload drifted from the additive key set: "
            f"{sorted(set(tick) ^ EXPECTED_STATUS_KEYS)}"
        )
    if terminal not in ticks:
        assert set(terminal) == EXPECTED_STATUS_KEYS, terminal

    if prelude_zero:
        for tick in ticks:
            if tick["state"] == "running" and tick["phase"] is None:
                # The prelude (the model-check + clone/pull span): the
                # scan has not started, so it has reported nothing. The
                # success terminal sets state BEFORE clearing phase, so
                # no terminal body can reach this arm.
                assert (tick["scan_done"], tick["scan_total"]) == (0, 0), (
                    f"the null prelude already carried scan counts: {tick}"
                )

    scan_ticks = [t for t in ticks if t["phase"] == "scan"]
    for tick in scan_ticks:
        assert tick["current_file"] is None, (
            f"the scan named a file — it counts, it does not read: {tick}"
        )
        assert tick["scan_total"] == scan_total, tick
        assert 0 < tick["scan_done"] <= scan_total, tick
    positions = [t["scan_done"] for t in scan_ticks]
    assert positions == sorted(positions), (
        f"the scan position went backwards across ticks: {positions}"
    )
    counted = [i for i, t in enumerate(ticks) if t["files_total"] > 0]
    if scan_ticks and counted:
        last_scan = max(i for i, t in enumerate(ticks) if t["phase"] == "scan")
        assert last_scan < counted[0], (
            "a scan frame followed a frame that already counted files — "
            "the phase machine ran backwards"
        )

    assert terminal["phase"] is None, terminal
    assert terminal["scan_total"] == scan_total, terminal
    assert terminal["scan_done"] == scan_total, (
        f"the terminal did not keep the completed scan: {terminal}"
    )
    assert terminal["files_total"] == files_total, terminal
    assert terminal["files_total"] <= terminal["scan_total"], (
        "the predicted change set is bigger than the scan that predicted "
        f"it: {terminal}"
    )


def _admin_sources_page(page: Page, app_url: str, *, settled: bool = True) -> None:
    """Real form login landing on the RAG page (the button's home).

    ``settled=True`` also waits for the boot re-attach to finish in a
    NON-running state (the phase-64 pattern — no later re-render can
    clobber the state under test). The served-status arms pass
    ``settled=False``: their boot fetch IS the state under test."""
    login(page, app_url)
    expect(page).to_have_url(app_url + SOURCES_URL, timeout=30_000)
    expect(page.locator("#sync-btn")).to_be_visible(timeout=30_000)
    if settled:
        expect(page.locator("#sync-label")).to_have_text(
            re.compile(r"^(Sync sources|Synced \d{1,2}:\d{2})$"), timeout=30_000
        )


# ---------------------------------------------------------------------------
# 1. (a) The count-down across a REAL sync of a REAL git source
# ---------------------------------------------------------------------------


def test_running_label_counts_down_across_a_real_sync(
    page: Page, app_url: str, db_ready: None, remaining_repo: Path
) -> None:
    """The owner-visible contract: the label counts DOWN and goes quiet
    at the final file — over a real ``file://`` git sync with the real
    (untouched) status endpoint."""
    page.set_default_timeout(SETTLE_TIMEOUT_MS)
    _admin_sources_page(page, app_url)
    _install_recorder(page)

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

    # The click's own state (the phase-64 contract the pre-walk keeps):
    # the button becomes the spinner and the box reveals the BARE label.
    expect(btn).to_be_disabled()
    expect(btn).to_have_attribute("aria-busy", "true")
    expect(page.locator("#sync-progress")).to_have_text("Syncing…")

    # The first file's window: the count part AND the chunk part (the
    # first file is the multi-chunk one, so its embed batches climb
    # while the count reads N − 1 remaining — the coexistence pin).
    expect(page.locator("#sync-progress-text")).to_have_text(
        re.compile(
            rf"^Syncing… {re.escape(SOURCE)}/{re.escape(FIRST_REL)}"
            rf" {MAX_COUNT} remaining · chunk \d+/\d+$"
        ),
        timeout=SETTLE_TIMEOUT_MS,
    )

    # The final file's window: the main part names it and NO count part
    # rides along (A1's hidden edge; the chunk part may — the last file
    # is multi-chunk too).
    expect(page.locator("#sync-progress-text")).to_have_text(
        re.compile(
            rf"^Syncing… {re.escape(SOURCE)}/{re.escape(LAST_REL)}"
            r"(?!.* remaining)(?: · chunk \d+/\d+)?$"
        ),
        timeout=SETTLE_TIMEOUT_MS,
    )

    terminal = recorder.stop()
    assert terminal["state"] == "success", terminal

    frames = _frames(page)
    texts = _texts(frames)

    # --- the rendered sequence's shape ---------------------------------
    import_frames = [f for f in frames if f["text"].startswith("Syncing… ")]
    assert import_frames, f"no running import label was ever rendered: {texts[:8]}"
    for frame in import_frames:
        assert IMPORT_LABEL_RE.match(frame["text"]), frame
        # The pre-phase position shape is gone from this part for good.
        assert not OLD_FRACTION_RE.search(frame["text"]), frame
        # A4: the untruncated label rides the title AND the announcer.
        assert frame["title"] == frame["text"], frame
        assert frame["aria"] == frame["text"], frame
        # The count span mirrors the part: visible iff the text carries it.
        assert frame["countHidden"] is not (" remaining" in frame["text"]), frame
        if " remaining" in frame["text"]:
            shown = re.search(r" (\d+) remaining", frame["text"])
            assert shown is not None, frame
            assert 1 <= int(shown.group(1)) <= MAX_COUNT, (
                f"a count-bearing label showed {shown.group(1)} "
                f"(allowed 1…{MAX_COUNT}): {frame}"
            )

    # --- the count-down itself -----------------------------------------
    observed: list[int] = []
    for label in texts:
        m = re.search(r" (\d+) remaining(?: · |$)", label)
        if m:
            value = int(m.group(1))
            if value not in observed:
                observed.append(value)
    assert len(observed) >= 3, (
        f"the label never visibly counted down (distinct N values: {observed})"
    )
    assert observed == sorted(observed, reverse=True), (
        f"the count part did not count DOWN: {observed}"
    )
    assert observed[0] == MAX_COUNT, f"the walk did not start at N − 1: {observed}"
    assert observed[-1] == 1, f"the count part never reached the last file: {observed}"
    assert 0 not in observed, "a '0 remaining' label rendered (A1's hidden edge)"

    # --- the chunk part coexists (phase 129, unbroken) ------------------
    coexist = [
        t for t in texts if re.search(r" \d+ remaining · chunk \d+/\d+$", t)
    ]
    assert coexist, f"the count and chunk parts never shared a label: {texts[:12]}"
    chunk_totals = {
        int(m.group(1))
        for t in texts
        for m in [re.search(r" · chunk \d+/(\d+)$", t)]
        if m
    }
    assert _expected_content_chunks(FIRST_REL) in chunk_totals, (
        f"the chunk part never reported the first file's real chunk count "
        f"({_expected_content_chunks(FIRST_REL)}): {sorted(chunk_totals)}"
    )

    # --- the A1 edge, sequence-wide ------------------------------------
    first_last = next(
        (i for i, t in enumerate(texts) if t.startswith(f"Syncing… {SOURCE}/{LAST_REL}")),
        None,
    )
    assert first_last is not None, f"the final file was never named: {texts[:12]}"
    after = texts[first_last:]
    assert not any(" remaining" in t for t in after), (
        f"the count part came back after the final file: {after[:8]}"
    )

    # --- the untouched phases and the untouched terminal line -----------
    assert any(t == "Writing KB overview…" for t in texts), texts[:20]
    summaries = [t for t in texts if t.startswith("Summarizing folders…")]
    assert summaries and all(OLD_FRACTION_RE.search(t) for t in summaries), (
        f"the summaries phase's folder position changed shape: {summaries}"
    )
    box = page.locator("#sync-progress-text")
    expect(box).to_have_text(counts_line(f"{N_FILES} added"))
    expect(page.locator("#sync-label")).to_have_text(SYNCED_LABEL)
    expect(btn).to_be_enabled()
    assert terminal["detail"]["added"] == N_FILES, terminal

    # --- the cross-layer invariant: the label is the payload's subtraction
    # Neither half of this changed in phase 194: the SUBTRACTION is the
    # phase-193 builder, and on this truncated-KB fixture the pair it
    # subtracts (the pre-scan's pending position, phase 194) coincides
    # with the walk position, so every rendered N still equals
    # total − done of the file the label named.
    position: dict[str, tuple[int, int]] = {}
    for tick in recorder.running_ticks:
        if tick["current_file"]:
            position[tick["current_file"]] = (tick["files_done"], tick["files_total"])
    assert {f"{SOURCE}/{rel}" for rel in REL_ORDER} <= set(position), (
        f"the status never reported every file: {sorted(position)}"
    )
    for frame in import_frames:
        named = frame["text"].split(" · ")[0]
        path = named[len("Syncing… ") :].split(" ")[0]
        done, total = position[path]
        assert total == N_FILES, (path, done, total)
        expected = total - done
        shown = re.search(r" (\d+) remaining", frame["text"])
        if expected > 0:
            assert shown is not None, (
                f"{path}: the payload said {expected} left, the label said none: {frame}"
            )
            assert int(shown.group(1)) == expected, (path, done, total, frame)
        else:
            assert shown is None, f"{path}: done==total but a part rendered: {frame}"

    # --- phase 194 (task 04, D5): the machine, off the SAME tick series --
    # The additive key set, the 0/0 prelude, the file-less scan ticks,
    # the one-directional sequence and the terminal that keeps BOTH
    # pairs (here equal, because a fresh KB's prediction IS its walk).
    _assert_phase_machine_tick_series(
        recorder.ticks, terminal, scan_total=N_FILES, files_total=N_FILES
    )

    # --- phase 194: the scan label, WHEREVER the page caught the stage ---
    # Whether the UI's 2 s poll lands inside a five-file scan is a coin
    # flip (test 2 pins the stage deterministically through a served
    # status); what is never a coin flip is the SHAPE of the frames it
    # does catch and where they may sit in the sequence.
    scan_frames = [f for f in frames if f["text"].startswith("Scanning…")]
    for frame in scan_frames:
        assert SCAN_LABEL_RE.fullmatch(frame["text"]), frame
        assert " remaining" not in frame["text"], frame
        assert SOURCE not in frame["text"], f"the scan named a file: {frame}"
        # A4: the untruncated label rides the title AND the announcer.
        assert frame["title"] == frame["text"], frame
        assert frame["aria"] == frame["text"], frame
        assert frame["countHidden"] is not ("(" in frame["text"]), frame
    first_import = next(
        (i for i, t in enumerate(texts) if t.startswith("Syncing… ")), None
    )
    if first_import is not None:
        assert all(
            i < first_import for i, t in enumerate(texts) if t.startswith("Scanning…")
        ), f"a scan label rendered after the import phase began: {texts[:14]}"


# ---------------------------------------------------------------------------
# 2. (c) The zero-total edges + the A1 hidden edge, on the exact payloads
#    the guards read
# ---------------------------------------------------------------------------


def test_prewalk_and_final_file_labels_have_no_count_part(
    page: Page, app_url: str, db_ready: None
) -> None:
    """The zero-total edges of A1 plus the phase-194 scan stage, pinned on
    the payloads the guards actually read (the served bodies carry the
    status endpoint's key set — what is under test is the client's ONE
    builder): ``scan_total === 0`` renders no count part, ``total === 0``
    renders none either, and neither does ``done === total``."""
    page.set_default_timeout(SETTLE_TIMEOUT_MS)
    stages: dict[str, Any] = {
        # The stage phase 194 inserts where the pre-walk used to stand:
        # the pre-scan mid-flight, denominator known, position climbing.
        "sync": _sync_status(
            state="running", phase="scan", scan_done=2, scan_total=N_FILES
        ),
        "upload": None,
    }
    _serve_statuses(page, stages)
    # settled=False: the boot fetch is the scan stub — the re-attach
    # itself is part of what this arm pins.
    _admin_sources_page(page, app_url, settled=False)
    _install_recorder(page)

    box = page.locator("#sync-progress-text")
    btn = page.locator("#sync-btn")

    # (phase 194, D4) The scan stage speaks its OWN label: "Scanning…"
    # plus the (n/total) part in the SAME count slot the import phase's
    # "N remaining" uses (one writer, one hidden-toggle, one A4
    # placement) — never a file, never a chunk part, and the full label
    # on the button title.
    expect(box).to_have_text(f"Scanning… (2/{N_FILES})", timeout=SETTLE_TIMEOUT_MS)
    # The PART's own toggle, not Playwright visibility: while a run is in
    # flight the button is the spinner and CSS hides the label spans, so
    # the honest pin is the writer's ``hidden`` flag plus the span's text
    # (the in-browser recorder in test 1 reads the same property).
    expect(page.locator(".sync-label-count")).to_have_text(f"(2/{N_FILES})")
    expect(page.locator(".sync-label-count")).to_have_js_property("hidden", False)
    expect(btn).to_have_attribute("title", f"Scanning… (2/{N_FILES})")
    expect(btn).to_be_disabled()

    # …and the same stage with NO denominator yet — the empty KB, and the
    # scan's own first frame: the phase-64 ``total === 0`` guard renders
    # the bare phase name, never "(0/0)".
    stages["sync"] = _sync_status(state="running", phase="scan")
    expect(box).to_have_text("Scanning…", timeout=SETTLE_TIMEOUT_MS)
    expect(page.locator(".sync-label-count")).to_have_text("")
    expect(page.locator(".sync-label-count")).to_have_js_property("hidden", True)
    expect(btn).to_have_attribute("title", "Scanning…")

    # (c) The null prelude — phase null, current_file null, files_total 0
    # (the pre-phase pre-walk, and exactly what the
    # BOR_SYNC_PENDING_COUNT=0 rollback reports for the whole run): the
    # bare prefix, no count part, the count span hidden, the title
    # identical (A4).
    stages["sync"] = _sync_status(state="running")
    expect(box).to_have_text("Syncing…", timeout=SETTLE_TIMEOUT_MS)
    expect(page.locator(".sync-label-count")).to_be_hidden()
    expect(btn).to_have_attribute("title", "Syncing…")
    expect(btn).to_be_disabled()

    # The walk knows the total: the first file counts down from N − 1.
    stages["sync"] = _sync_status(
        state="running",
        phase="import",
        current_file=f"{SOURCE}/{FIRST_REL}",
        files_done=1,
        files_total=N_FILES,
    )
    expect(box).to_have_text(
        f"Syncing… {SOURCE}/{FIRST_REL} {MAX_COUNT} remaining",
        timeout=SETTLE_TIMEOUT_MS,
    )

    # The last count-bearing position, with the phase-129 chunk part
    # beside it (the two parts are independent and coexist).
    stages["sync"] = _sync_status(
        state="running",
        phase="import",
        current_file=f"{SOURCE}/{REL_ORDER[3]}",
        files_done=N_FILES - 1,
        files_total=N_FILES,
        current_chunk=2,
        chunks_total=5,
    )
    expect(box).to_have_text(
        f"Syncing… {SOURCE}/{REL_ORDER[3]} 1 remaining · chunk 2/5",
        timeout=SETTLE_TIMEOUT_MS,
    )

    # (a/A1) The final file: done == total → the part is ABSENT while the
    # main part still names the file (and the chunk part may stay).
    stages["sync"] = _sync_status(
        state="running",
        phase="import",
        current_file=f"{SOURCE}/{LAST_REL}",
        files_done=N_FILES,
        files_total=N_FILES,
        current_chunk=1,
        chunks_total=4,
    )
    expect(box).to_have_text(
        f"Syncing… {SOURCE}/{LAST_REL} · chunk 1/4",
        timeout=SETTLE_TIMEOUT_MS,
    )
    expect(page.locator(".sync-label-count")).to_be_hidden()
    expect(btn).to_have_attribute("title", f"Syncing… {SOURCE}/{LAST_REL} · chunk 1/4")

    # The success terminal: the untouched phase-64 counts line. The
    # stamp is NOW (the phase-154 rule renders a same-day completion as
    # the plain "Synced HH:MM").
    stages["sync"] = _sync_status(
        state="success",
        started_at=_now_iso(),
        finished_at=_now_iso(),
        files_done=N_FILES,
        files_total=N_FILES,
        # The scan pair KEPT at the terminal beside the file counts (the
        # phase-64 keep-final-values convention) — the counts line below
        # ignores them, which is the point of pinning them here.
        scan_done=N_FILES,
        scan_total=N_FILES,
        detail={"added": N_FILES, "updated": 0, "unchanged": 0, "pruned": 0},
    )
    expect(box).to_have_text(counts_line(f"{N_FILES} added"), timeout=SETTLE_TIMEOUT_MS)
    expect(page.locator("#sync-label")).to_have_text(SYNCED_LABEL, timeout=30_000)
    expect(btn).to_be_enabled()

    # Nothing rendered in this run ever spoke the pre-phase shape on a
    # "Syncing…" label or a "0 remaining".
    texts = _texts(_frames(page))
    assert not any(OLD_FRACTION_RE.search(t) for t in texts if t.startswith("Syncing…")), texts
    assert not any("0 remaining" in t for t in texts), texts
    assert "Syncing…" in texts, texts

    # The scan stage rendered, in its own shape, and ONLY before the
    # import phase started: the (n/m) is the scan's position (and the
    # summaries phase's), never the sync's.
    scan_texts = [t for t in texts if t.startswith("Scanning…")]
    assert f"Scanning… (2/{N_FILES})" in scan_texts, texts
    assert "Scanning…" in scan_texts, texts
    for label in scan_texts:
        assert SCAN_LABEL_RE.fullmatch(label), label
        assert " remaining" not in label, label
    first_import = next(
        (i for i, t in enumerate(texts) if t.startswith("Syncing… ")), None
    )
    assert first_import is not None, texts
    assert all(
        i < first_import for i, t in enumerate(texts) if t.startswith("Scanning…")
    ), texts


# ---------------------------------------------------------------------------
# 3. (b) A2 — the upload job shares the builder and the shape
# ---------------------------------------------------------------------------


def test_upload_label_shares_the_count_down_builder(
    page: Page, app_url: str, db_ready: None, upload_archive: Path
) -> None:
    """A2: the upload's running label is the SAME builder's output, so it
    counts down in the SAME shape."""
    page.set_default_timeout(SETTLE_TIMEOUT_MS)

    # The real upload run, against the live endpoint first: since phase
    # 90 it is unpack + register only, so its status carries
    # null/0/0 for the WHOLE run — the reason the shared shape is pinned
    # through a served status below (a real upload can never show a
    # count at all, and can therefore never regress to "(n/m)").
    with httpx.Client(base_url=app_url, timeout=60.0) as client:
        r = client.post("/api/login", json={"password": ADMIN_PASSWORD})
        assert r.status_code == 204, r.text
        with upload_archive.open("rb") as fh:
            r = client.post(
                "/api/git-sources/upload",
                files={"file": (f"{UPLOAD_NAME}.tar.gz", fh, "application/gzip")},
            )
        assert r.status_code == 202, r.text
        body: dict[str, Any] = {}
        for _ in range(200):
            body = client.get("/api/git-sources/upload/status").json()
            if body["state"] in ("success", "failed"):
                break
            time.sleep(0.05)
        assert body["state"] == "success", body
        assert body["current_file"] is None, body
        assert body["files_done"] == 0 and body["files_total"] == 0, body

    # The shared shape, on the page's own poll: a running upload status
    # that DOES carry counts (the key set the endpoint already answers
    # with) renders the count-down part through the one builder.
    stages: dict[str, Any] = {
        "sync": _sync_status(state="idle"),
        "upload": _upload_status(
            state="running",
            current_file=f"{UPLOAD_NAME}/docs/01.md",
            files_done=1,
            files_total=3,
        ),
    }
    _serve_statuses(page, stages)
    # settled=False: the boot re-attach of the in-flight UPLOAD run is
    # the entry point this arm pins.
    _admin_sources_page(page, app_url, settled=False)
    _install_recorder(page)

    box = page.locator("#sync-progress-text")
    btn = page.locator("#sync-btn")
    expect(box).to_have_text(
        f"Importing {UPLOAD_NAME}/docs/01.md 2 remaining", timeout=SETTLE_TIMEOUT_MS
    )
    expect(btn).to_have_attribute("title", f"Importing {UPLOAD_NAME}/docs/01.md 2 remaining")
    expect(btn).to_be_disabled()

    # The upload's final file: the same A1 edge through the same builder.
    stages["upload"] = _upload_status(
        state="running",
        current_file=f"{UPLOAD_NAME}/docs/02.md",
        files_done=3,
        files_total=3,
    )
    expect(box).to_have_text(f"Importing {UPLOAD_NAME}/docs/02.md", timeout=SETTLE_TIMEOUT_MS)
    expect(page.locator(".sync-label-count")).to_be_hidden()

    # The upload settle (branch 6): the button returns to "Sync sources"
    # and the box hides — the upload's own unchanged surface.
    stages["upload"] = _upload_status(
        state="success", detail={"message": "uploaded"}, finished_at=_now_iso()
    )
    expect(page.locator("#sync-label")).to_have_text("Sync sources", timeout=SETTLE_TIMEOUT_MS)
    expect(page.locator("#sync-progress")).to_be_hidden(timeout=SETTLE_TIMEOUT_MS)

    texts = _texts(_frames(page))
    importing = [t for t in texts if t.startswith("Importing")]
    assert importing, texts
    for label in importing:
        assert re.fullmatch(
            rf"Importing {re.escape(UPLOAD_NAME)}/docs/\d{{2}}\.md"
            r"(?: [12] remaining)?",
            label,
        ), label
        assert not OLD_FRACTION_RE.search(label), label
        # The upload branch never receives a status object (its endpoint
        # has no phase / chunk / image fields — phase 129 / 152), so the
        # chunk part belongs to the sync arms only.
        assert " · chunk " not in label, label


# ---------------------------------------------------------------------------
# 4. (phase 194, task 04 / D5) The re-sync: the prediction, not the walk
# ---------------------------------------------------------------------------


def test_resync_of_an_unchanged_repo_reports_zero_work(
    page: Page, app_url: str, db_ready: None, remaining_repo: Path
) -> None:
    """The run phase 193's fixture never had: the SAME repo synced twice.

    Run 1 fills the truncated KB, so its predicted change set IS the
    whole walk (``files_total == scan_total == N_FILES`` — the parity
    test 1 rides). Run 2 finds every byte where it left them, so the
    pre-scan predicts NOTHING: the ``progress`` pair is ``0/0`` on every
    tick and the client's ``total === 0`` guard (test 2) renders no count
    part — while ``scan_done`` / ``scan_total`` report the whole KB walked
    and are KEPT at the terminal beside the zeroed file pair. That
    divergence (5 scanned, 0 to re-index) is phase 194 in one line, and
    it is exactly the shape that used to render "``N`` remaining" for
    files nobody was going to re-index (the owner's prod report).

    A no-change run makes ZERO LLM call (the phase-147 fast path), so its
    import window is milliseconds and the 2 s UI poll is not expected to
    render a file label for it: the pin is the WIRE pair plus the
    ABSENCE of a count part in whatever the page did render, not a
    rendered per-file label (test 1 covers the rendered side).
    """
    page.set_default_timeout(SETTLE_TIMEOUT_MS)
    _admin_sources_page(page, app_url)
    _install_recorder(page)
    btn = page.locator("#sync-btn")
    box = page.locator("#sync-progress-text")

    # --- run 1: the fresh import, the parity test 1 rides ---------------
    expect(btn).to_be_enabled()
    # wait_for_running: test 1's TERMINAL is still what the endpoint
    # answers with here (the run state lives in the app, not the page),
    # and without the flag this recorder would hand that body back as its
    # own before this test had synced anything at all.
    rec1 = _TickRecorder(app_url, "/api/sync/status", wait_for_running=True)
    rec1.start()
    btn.click()
    terminal1 = rec1.stop()
    assert terminal1["state"] == "success", terminal1
    expect(box).to_have_text(
        counts_line(f"{N_FILES} added"), timeout=SETTLE_TIMEOUT_MS
    )
    assert terminal1["files_total"] == N_FILES, terminal1
    assert terminal1["scan_total"] == N_FILES, terminal1
    frame_mark = page.evaluate("() => window.__borFrames.length")

    # --- run 2: the same bytes, the owner's incremental case ------------
    expect(btn).to_be_enabled(timeout=SETTLE_TIMEOUT_MS)
    rec2 = _TickRecorder(app_url, "/api/sync/status", wait_for_running=True)
    rec2.start()
    btn.click()
    terminal2 = rec2.stop()
    assert terminal2["state"] == "success", terminal2

    # The owner-visible terminal line stays the byte-identical phase-64
    # counts line — the exact surface the prediction never replaces.
    expect(box).to_have_text(
        counts_line(f"0 added · {N_FILES} unchanged"), timeout=SETTLE_TIMEOUT_MS
    )
    expect(page.locator("#sync-label")).to_have_text(SYNCED_LABEL)
    expect(btn).to_be_enabled()
    detail = terminal2["detail"]
    assert (detail["added"], detail["updated"], detail["unchanged"]) == (
        0,
        0,
        N_FILES,
    ), detail

    # THE PHASE IN ONE LINE: the scan walked the KB, the work was nothing.
    assert terminal2["scan_total"] == N_FILES, terminal2
    assert terminal2["scan_done"] == N_FILES, terminal2
    assert terminal2["files_total"] == 0, (
        f"a no-change run reported a non-zero denominator: {terminal2}"
    )
    assert terminal2["files_done"] == 0, terminal2
    for tick in rec2.ticks:
        assert (tick["files_done"], tick["files_total"]) == (0, 0), (
            f"the empty pending set leaked a walk position: {tick}"
        )
        assert tick["scan_total"] in (0, N_FILES), tick
    # `prelude_zero=False`: the runner resets ``state`` to ``running``
    # one synchronous statement before it resets the scan pair, and the
    # status endpoint answers from a threadpool — a poll landing in that
    # window would legitimately see the PREVIOUS run's kept 5/5. The
    # claim itself is pinned race-free in test 1 and test 2; repeating it
    # here would buy nothing but a 1-in-100k flake.
    _assert_phase_machine_tick_series(
        rec2.ticks, terminal2, scan_total=N_FILES, files_total=0, prelude_zero=False
    )

    # Whatever run 2 rendered carried NO count part: no "N remaining"
    # (the empty pending set → the total === 0 guard) and no scan (n/m)
    # once the import phase had begun.
    run2_texts = _texts(_frames(page)[frame_mark:])
    assert run2_texts, (
        f"nothing was rendered after the run-1 settle mark ({frame_mark}) — "
        "the click's own label is missing from the recorder"
    )
    assert not any(" remaining" in t for t in run2_texts), run2_texts
    for label in run2_texts:
        if label.startswith("Scanning…"):
            assert SCAN_LABEL_RE.fullmatch(label), label
        if label.startswith("Syncing…"):
            assert not OLD_FRACTION_RE.search(label), label
    scan_idx = [i for i, t in enumerate(run2_texts) if t.startswith("Scanning…")]
    imp_idx = [i for i, t in enumerate(run2_texts) if t.startswith("Syncing… ")]
    if scan_idx and imp_idx:
        assert max(scan_idx) < imp_idx[0], run2_texts