"""Phase 187 (task 04) E2E (Playwright, mock-only): the sync's PEAK MEMORY
is visible on the page that started the run — and it is the truth.

Source: owner incident 2026-10-15. Reconstructing "how much RAM did that
sync use?" meant reading ``systemd`` unit-teardown lines from three boots
(``brain-of-code-app.service: … 3.3G memory peak``), and boot −1 — the run
that killed the box — had NO teardown record at all, because the VM never
shut down cleanly. The app had no opinion about its own memory: no log
line, no status field, nothing on the Sources page. Phase 187 made it a
number (``app/core/memory.py``, the lifecycle lines, the status keys) and
this module is the browser proof of the last mile.

What the phase put on the wire, in the operator's words:

* ``GET /api/sync/status`` carries the ADDITIVE pair ``rss_mb`` (the
  process's CURRENT resident set, MB) and ``rss_peak_mb`` (its HIGH-WATER
  mark) — ``int | null``, where ``null`` means "no reading" and NEVER 0;
  sampled at the run start, at every ``phase`` transition, throttled to
  one read per 5 s through the import's per-file hook, and FORCED in every
  terminal, which KEEPS the final pair next to the final counts;
* the Sources page appends `` · peak <N> MB`` to the result line it
  already spoke (the ``fmtSyncResult`` counts line, "Sync cancelled") —
  trailing text on a line that already exists, never a new control, never
  a new live region, and NOTHING rendered when the pair is null.

**Which number this is (read it before comparing it to a dashboard).**
Both fields are PROCESS readings — ``VmRSS`` from ``/proc/self/status`` and
``getrusage(RUSAGE_SELF).ru_maxrss`` — NOT the host's cgroup figure. A
cgroup counts PAGE CACHE too: the incident box's app cgroup showed ~2.16 GB
of ``file`` against ~211 MB of ``anon`` while idle, and ``systemd`` printed
``3.3G memory peak`` for the unit while the uvicorn process was a fraction
of that. Neither number is wrong; LOCKED A1 pins the app to the process
one, and `README.md` (*Operational notes: the sync's memory (phase 187)*)
owns the full caveat and the operator grep
(``grep ' :: ' server.log | grep ' rss_peak_mb='``). This module never
asserts a MAGNITUDE — only that the number is present, typed, terminal,
honest (``peak >= current``, with the one exception the sampler owns: a
kernel high-water refreshed lazily is lifted to the current reading, never
invented from it), and equal on both surfaces.

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

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

MOCK-ONLY suite: ``E2E_REAL_LLM=1`` is not supported — the deterministic
mock plus the phase-64 ``tests/e2e/slow_llm.py`` delay proxy is what makes
the RUNNING observation a fact instead of a race: the run must outlive the
UI's 2 s status poll (and the tests' ~100 ms one) so the "no clause while
running" pin and the cancel window are both deterministic. Two files →
≈ 10 LLM calls (the two model probes, the per-file embed batches + the
summaries, the KB overview, the source-root folder summary) ≈ 5 s at
``SLOW_DELAY_S = 0.5``.

Isolation (the house shape): ONE module app on its own port (the conftest
session app owns 8123 in a combined run), the mock LLM on the conftest
port behind the proxy, a dedicated ``BOR_SOURCES_DIR`` /
``BOR_PDF_PAGE_DIR`` so no run writes into the operator's, the shared E2E
Postgres truncated BEFORE **and** AFTER every test (after waiting for the
app to hold no running sync), and the admin in through the real form login
(``tests/e2e/auth_helpers.py``). Each test seeds the same two-file
``kind=local`` source directory and registers it through the authenticated
API, so any single test re-runs alone with a fresh KB and the counts line
says ``2 added``.

Test → observable mapping (Playwright Mapping Rule — the rendered DOM is
the observable; the DOM number and the payload number are read as one
comparison, so "the page lies" cannot slip through):

1. ``test_result_line_shows_the_peak`` — a full run from the Sources page:
   the box (``#sync-progress-text``) settles the counts line WITH the
   trailing clause (``^2 added · peak \\d+ MB$``), the clause matches
   ``/peak \\d+ MB/``, the aria-live region IS the box (``role=status`` +
   ``aria-live=polite``, no ``aria-hidden``) and carries the SAME text,
   the clause is TEXT and not an element (the box still holds exactly one
   child span — no new control), and the megabytes in the DOM EQUAL the
   terminal payload's ``rss_peak_mb``. Where the platform cannot know the
   pair, the mirror contract is pinned instead: NO clause at all.
2. ``test_status_payload_reports_rss`` — the endpoint end to end: a tick
   during ``running`` (a) reports the pair with the documented types
   (``int | null``), (b) carries the pre-phase key set UNCHANGED (the
   additive pin: exactly the pre-187 keys plus the memory pair) and (c)
   shows NO ``peak`` clause on the page — the clause is terminal-only;
   after the terminal ``rss_peak_mb`` is a POSITIVE int with
   ``peak >= current`` (the invariant the ONE sampler owns: the kernel
   refreshes its high-water lazily, so ``_sample_rss`` lifts a lagging peak
   to the current reading — a peak under the run's own footprint would be a
   false number), and the same pair survives a later poll (the page
   reloaded an hour later still answers).
3. ``test_anonymous_cannot_read_memory`` — the A4 exposure guard: a
   logged-out ``GET /api/sync/status`` is 403 with the pre-phase body
   ``{"detail": "admin only"}`` — no memory number on an anonymous
   surface, before or after a run — and the page shows the sign-in gate
   with no sync button and no peak clause anywhere in the DOM.
4. ``test_cancelled_terminal_keeps_the_peak`` — the ``test_sync_cancel.py``
   shape: cancel mid-run and the NEUTRAL "Sync cancelled" state still
   renders (no banner, no modal, the button retry-ready) with the clause
   present and equal to the cancelled payload's ``rss_peak_mb`` — the
   run that stopped early is exactly the run whose memory matters.
5. ``test_peak_clause_fits_at_375`` — PLAN §7 at phone width: after a run
   settled at 375 px, the clause is on screen, the box does not clip it
   (``scrollWidth <= clientWidth`` for the box and its text), the box
   stays inside the viewport and the PAGE has no horizontal overflow.
"""
from __future__ import annotations

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

import httpx
import pytest
from playwright.sync_api import Browser, ConsoleMessage, Page, ViewportSize, expect
from sqlalchemy import text

from app.config import Settings as _Settings
from app.core.memory import rss_mb, rss_peak_mb
from app.db import SessionLocal
from e2e.auth_helpers import login
from e2e.conftest import (
    ADMIN_PASSWORD,
    MOCK_PORT,
    SESSION_SECRET,
    USE_REAL_LLM,
    _wait_http,
)

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

#: Dedicated module ports (the house env-overridable pattern): the conftest
#: session app owns 8123 and the mock owns 8901 in a combined run.
APP_PORT = int(os.environ.get("E2E_APP_PORT_PEAKMEM", "8204"))
APP_URL = f"http://127.0.0.1:{APP_PORT}"

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

#: Per-LLM-request delay (the phase-64 timing convention): the run spans
#: ≈ 10 calls ≈ 5 s, i.e. ≫ the UI's 2 s status poll — the running window
#: and the cancel window are both observable, not races.
SLOW_DELAY_S = "0.5"

SOURCES_URL = "/sources.html"

#: The fixture directory's basename — the ``source`` a ``kind=local`` row
#: records (``Path(path).name``), stable whatever temp root pytest picked.
#:
#: **Nothing here is allowed to contain the word "peak"** — not the source
#: name, not a path, no marker. Several pins below ask whether the page
#: says ANYTHING about memory (the running label must be clause-free, the
#: anonymous shell must be silent), and those asks read the rendered text;
#: a fixture path like ``peakmem/notes/peak-alpha.md`` rides the running
#: label verbatim and would answer "yes" from the file name alone. So the
#: fixture is ``memline-fixture/notes/mem-*.md`` and the clause stays the
#: only thing on the page that can say the word.
SOURCE_NAME = "memline-fixture"
N_FILES = 2
FIXTURE_FILES = {
    "notes/mem-alpha.md": (
        "# Memory line alpha\n"
        "\n"
        "Head marker RESE-MEMLINE-ALPHA-6d13 opens this note.\n"
        "\n"
        "## Alpha body\n"
        "\n"
        "A small note that exists so the phase-187 suite has a real run "
        "to watch: a real walk, real embeds, a real KB overview, and the "
        "run's resident-set clause on the line it settles.\n"
    ),
    "notes/mem-beta.md": (
        "# Memory line beta\n"
        "\n"
        "Head marker RESE-MEMLINE-BETA-9a74 opens this note — the second "
        "file, the counts line's denominator.\n"
    ),
}

#: The clause sources.js appends (``fmtSyncPeakClause``): trailing text on
#: a result line that already exists. REQUIRED here — this suite's whole
#: point is that the number is there (the optional form, for the suites
#: that only pin their own counts, lives in ``e2e.sync_helpers``).
PEAK_CLAUSE = re.compile(r" · peak (\d+) MB$")

#: The clause SHAPE, for the "this surface says nothing about memory" pins
#: (the running label, the anonymous shell). Matched case-insensitively on
#: the word, because the clause is the only thing on these pages allowed to
#: use it — which the fixture naming rule above is what keeps true.
_ANY_PEAK = re.compile(r"peak", re.IGNORECASE)

#: The settled success line: the pre-phase ``fmtSyncResult`` counts line
#: (a fresh KB of two files → "2 added", the zero "updated"/"unchanged"
#: parts omitted) with the phase-187 clause trailing it.
SUCCESS_LINE = re.compile(rf"^{N_FILES} added{PEAK_CLAUSE.pattern}$")

#: The neutral cancel line (phase 134/150 wording, byte-identical) with
#: the clause trailing it.
CANCELLED_LINE = re.compile(rf"^Sync cancelled{PEAK_CLAUSE.pattern}$")

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

#: The status payload's key set BEFORE phase 187 — the additive pin's
#: left-hand side. The payload must stay exactly this set plus the memory
#: pair, forever (the same list the integration suite pins).
PRE_187_STATUS_KEYS = frozenset(
    {
        "state",
        "started_at",
        "finished_at",
        "detail",
        "error",
        "current_file",
        "files_done",
        "files_total",
        "current_chunk",
        "chunks_total",
        "images_done",
        "images_total",
        "phase",
        "current_summary",
        "summaries_done",
        "summaries_total",
    }
)
MEMORY_KEYS = frozenset({"rss_mb", "rss_peak_mb"})

#: The pre-phase integer count slots (their types are part of the
#: unchanged-shape pin in test 2 — never a float, never null).
_COUNT_KEYS = frozenset(
    {
        "files_done",
        "files_total",
        "current_chunk",
        "chunks_total",
        "images_done",
        "images_total",
        "summaries_done",
        "summaries_total",
    }
)

#: The invariant the page can rely on: ``peak >= current``. The raw kernel
#: counters do NOT give that for free (``VmRSS`` and ``ru_maxrss`` are two
#: different sources, both floored to whole MB, and the kernel refreshes the
#: high-water lazily — measured at ``rss_mb=164`` against
#: ``rss_peak_mb=162``), so the ONE sampling point
#: (:func:`app.api.sync._sample_rss`) lifts a lagging peak to the current
#: reading. Slack ZERO: a peak BELOW the run's own current footprint is a
#: false number, and the page must never show one.
MB_SLACK = 0

#: Does THIS platform answer the two helpers? Linux (the deploy target:
#: ``/proc`` + ``resource``) always does. The app under test is a
#: subprocess of this process's own OS, so this answer is the app's answer.
#: Where it is False the value pins step aside and the ABSENCE contract is
#: pinned instead (null renders nothing — never "peak ? MB", never 0).
MEMORY_KNOWABLE = rss_mb() is not None and rss_peak_mb() is not None

#: The house phone viewport at which the clause must still fit (the
#: PLAN §7 / AGENTS rule 5 width: no horizontal overflow at 375 px).
MOBILE_VIEWPORT: ViewportSize = {"width": 375, "height": 720}

#: Settle budgets: a two-file run under the proxy is ≈ 5 s and the UI's 2 s
#: poll settles at most one tick after the terminal — 90 s is an order of
#: magnitude of head-room (the house sync budget, no client timeout).
SETTLE_TIMEOUT_MS = 90_000

#: The browser logs "Failed to load resource: … 403" for the anonymous
#: probes (the whoami gate + the admin-only status fetch) — expected
#: noise, not a defect.
_CONSOLE_NOISE = re.compile(r"Failed to load resource", re.IGNORECASE)


# ---------------------------------------------------------------------------
# The fixture source
# ---------------------------------------------------------------------------


def _seed_source(root: Path) -> Path:
    """(Re-)write the two fixture files under ``root / SOURCE_NAME``.

    Idempotent on purpose: every test calls it, so the tree is at its
    ORIGINAL bytes whenever a test starts and a lone re-run of any single
    test still sees a fresh KB with ``2 added``.
    """
    base = root / SOURCE_NAME
    for rel, body in FIXTURE_FILES.items():
        target = base / rel
        target.parent.mkdir(parents=True, exist_ok=True)
        target.write_text(body, encoding="utf-8")
    assert not (base / ".git").exists()  # a plain directory — no git anywhere
    return base


@pytest.fixture(scope="module")
def source_dir(tmp_path_factory: pytest.TempPathFactory) -> Path:
    """The fixture tree the module app walks (the app server runs on the
    same host, so the path is visible to it)."""
    return _seed_source(tmp_path_factory.mktemp("bor_peak_memory"))


# ---------------------------------------------------------------------------
# The module app + the timing proxy
# ---------------------------------------------------------------------------


@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 ONLY thing that makes the RUNNING
    observation and the cancel window deterministic (module docstring)."""
    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,
    source_dir: Path,
    tmp_path_factory: pytest.TempPathFactory,
) -> Iterator[str]:
    """The real app under test — per-module env (the conftest pattern):
    ``BOR_LLM_BASE_URL`` is the SLOW proxy in front of the deterministic
    mock, ``BOR_GIT_SOURCES`` forced empty (each test registers its own
    local row; an operator's gitignored ``.env`` must never leak a real
    source into a run), and a dedicated ``BOR_SOURCES_DIR``.
    """
    if USE_REAL_LLM:
        pytest.skip(
            "mock-only suite — the running window IS the slow proxy in "
            "front of the deterministic mock (E2E_REAL_LLM unsupported)"
        )
    assert mock_llm and source_dir
    scratch = tmp_path_factory.mktemp("bor_peak_memory_app")
    env = dict(os.environ)
    env.pop("DEBUGPY", None)
    env["BOR_ENVIRONMENT"] = "e2e"
    env["BOR_STATIC_DIR"] = str(REPO / "frontend")
    env["BOR_LLM_BASE_URL"] = f"{SLOW_URL}/v1"
    # Mock-calibrated gate (the conftest pattern): both floors must stay
    # <= the threshold or the app refuses to boot (app/config.py).
    env["BOR_RELEVANCE_THRESHOLD"] = "0.30"
    env["BOR_LEXICAL_SUPPORT_FLOOR"] = "0.15"
    env["BOR_SOURCE_USEFULNESS_FLOOR"] = "0.15"
    # Instant retry waits + the code-default budget (the conftest leak
    # guard — this suite never needs a retry, and never waits 5 s for one).
    env["BOR_LLM_RETRY_DELAY"] = "0"
    env["BOR_LLM_RETRIES"] = str(_Settings.model_fields["llm_retries"].default)
    env.setdefault(
        "BOR_DATABASE_URL",
        "postgresql+psycopg://reese:reese@localhost:5432/brain_of_reese",
    )
    # Phase 16: admin auth must be set or create_app() refuses to boot.
    env["BOR_ADMIN_PASSWORD"] = ADMIN_PASSWORD
    env["BOR_SESSION_SECRET"] = SESSION_SECRET
    env["BOR_GIT_SOURCES"] = ""
    env["BOR_SOURCES_DIR"] = str(scratch / "checkouts")
    env["BOR_PDF_PAGE_DIR"] = str(scratch / "pdf-pages")
    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


# ---------------------------------------------------------------------------
# Isolation
# ---------------------------------------------------------------------------


def _truncate_all() -> None:
    """Fresh registry + KB (the E2E isolation pattern, the phase-184 list:
    ``sources_meta`` and ``folder_summaries`` reset with the KB)."""
    with SessionLocal() as db:
        db.execute(
            text(
                "TRUNCATE chunks, documents, query_log, kb_overview, "
                "git_sources, sources_meta, folder_summaries"
            )
        )
        db.commit()  # without the commit the TRUNCATE rolls back


def _wait_sync_settled(base_url: str, timeout_s: float = 180.0) -> None:
    """No run may leak across a truncate: wait for a non-running status with
    an own admin session (the endpoint is admin-only). Usually a no-op —
    every test settles before it ends."""
    deadline = time.monotonic() + timeout_s
    while time.monotonic() < deadline:
        with httpx.Client(base_url=base_url, timeout=5.0) as client:
            client.post("/api/login", json={"password": ADMIN_PASSWORD})
            r = client.get("/api/sync/status")
            assert r.status_code == 200, r.text
            if r.json()["state"] != "running":
                return
        time.sleep(0.2)
    raise AssertionError(f"a sync was still running at a test boundary: {base_url}")


@pytest.fixture(autouse=True)
def _clean(app_url: str, source_dir: Path, db_ready: None) -> Iterator[None]:
    """Per-test KB isolation: the shared E2E Postgres is truncated before
    AND after every test, the app holding no running sync at both
    boundaries. The fixture tree is re-seeded first (idempotent) and the
    local row is registered through the authenticated API, so every test
    owns a fresh two-file source and its own ``2 added``."""
    _wait_sync_settled(app_url)
    _truncate_all()
    _seed_source(source_dir.parent)
    with httpx.Client(base_url=app_url, timeout=30.0) as client:
        assert client.post("/api/login", json={"password": ADMIN_PASSWORD}).status_code == 204
        r = client.post(
            "/api/git-sources", json={"kind": "local", "path": str(source_dir)}
        )
        assert r.status_code == 201, r.text
    yield
    _wait_sync_settled(app_url)
    _truncate_all()


@pytest.fixture()
def page(browser: Browser) -> Iterator[Page]:
    """The conftest's page + the console-error collector: an uncaught JS
    exception (pageerror) or a console error that is not the browser's
    expected 403 "Failed to load resource" noise fails the owning test at
    teardown — a clause builder that threw inside ``setSyncProgress`` would
    otherwise settle a blank box and read as a passing test.

    The conftest's ``page`` is NOT reused here because this suite's whole
    subject is the sync result line: a JS error inside the clause builder
    must fail the owning test, not leak (the ``test_sync_cancel.py``
    pattern)."""
    pg = browser.new_page(viewport={"width": 1280, "height": 800})
    errors: list[str] = []

    def _on_console(msg: ConsoleMessage) -> None:
        if msg.type == "error" and not _CONSOLE_NOISE.search(msg.text):
            errors.append(msg.text)

    pg.on("console", _on_console)
    pg.on("pageerror", lambda err: errors.append(str(err)))
    try:
        yield pg
    finally:
        pg.close()
        assert not errors, f"console/page errors: {errors}"


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


def _status(page: Page, base_url: str) -> dict[str, Any]:
    """The status payload through the ADMIN browser context (the cookie
    rides ``page.request`` — the same surface the page's own poll uses)."""
    r = page.request.get(f"{base_url}/api/sync/status")
    assert r.status == 200, r.text
    body: dict[str, Any] = r.json()
    return body


def _wait_sync_state(
    page: Page, base_url: str, states: set[str], timeout_s: float = 90.0
) -> dict[str, Any]:
    """Poll the status endpoint at ~100 ms until the run's state is in
    *states* (the deterministic layer: the UI's own poll ticks at 2 s)."""
    deadline = time.monotonic() + timeout_s
    body: dict[str, Any] = {}
    while time.monotonic() < deadline:
        body = _status(page, base_url)
        if body["state"] in states:
            return body
        time.sleep(0.1)
    raise AssertionError(f"sync did not reach {sorted(states)}: {body}")


def _open_sources(page: Page, base_url: str) -> None:
    """The real form login, landing on the Sources view (the sync button's
    home) with the module graph booted."""
    login(page, base_url)  # lands on /sources.html
    btn = page.locator("#sync-btn")
    expect(btn).to_be_visible(timeout=30_000)
    expect(btn).to_be_enabled()


def _click_sync(page: Page) -> None:
    """Click **Sync sources** and pin the 202 moment: the button disables
    (the run is in flight), the box reveals with the bare running label —
    which carries NO peak clause (the clause is terminal-only)."""
    btn = page.locator("#sync-btn")
    btn.click()
    expect(btn).to_be_disabled()
    expect(page.locator("#sync-progress")).to_have_text("Syncing…")


def _start_from_page(page: Page, base_url: str) -> None:
    """Login on the Sources view, then the click (the two-step form of the
    same helper, so a test can change the viewport BETWEEN them)."""
    _open_sources(page, base_url)
    _click_sync(page)


def _clause_number(text: str) -> int:
    """The MB value the clause carries (loudly absent when there is none)."""
    m = PEAK_CLAUSE.search(text)
    assert m, f"no peak clause in {text!r}"
    return int(m.group(1))


# ---------------------------------------------------------------------------
# 1. The success terminal: the clause on the line that already existed
# ---------------------------------------------------------------------------


def test_result_line_shows_the_peak(page: Page, app_url: str, db_ready: None) -> None:
    """A real sync started from the Sources page ends with the number on
    the line the page already spoke — ``2 added · peak <N> MB`` — inside
    the sole aria-live region, as TEXT (no new control), and equal to the
    terminal payload's ``rss_peak_mb`` (the page and the log can never
    disagree, because they read the one forced terminal sample)."""
    page.set_default_timeout(30_000)
    _start_from_page(page, app_url)

    terminal = _wait_sync_state(page, app_url, {"success", "failed", "cancelled"})
    assert terminal["state"] == "success", terminal
    assert terminal["error"] is None, terminal

    # The UI settle (the page's own 2 s poll, one tick after the terminal).
    expect(page.locator("#sync-label")).to_have_text(SYNCED_LABEL, timeout=SETTLE_TIMEOUT_MS)
    box = page.locator("#sync-progress")
    expect(box).to_be_visible()
    text_el = page.locator("#sync-progress-text")
    line = text_el.inner_text()

    if not MEMORY_KNOWABLE:
        # The mirror contract: no reading → NO clause at all (never
        # "peak ? MB", never "peak 0 MB"), and the counts line alone.
        assert line == f"{N_FILES} added", line
        assert not _ANY_PEAK.search(box.inner_text()), box.inner_text()
        return

    # The clause is there AND the pre-phase wording is intact (the counts
    # line keeps its exact bytes; the clause trails it).
    assert SUCCESS_LINE.fullmatch(line), line
    assert PEAK_CLAUSE.search(line), line
    assert line.startswith(f"{N_FILES} added"), line

    # The clause is TEXT on a line that already exists — the box still
    # holds its ONE child span (no new control, no new live region).
    expect(box.locator("> *")).to_have_count(1)

    # The box IS the sole aria-live region and it announces the clause
    # (phase 150: one live region, one surface — and the clause rides it).
    expect(box).to_have_attribute("role", "status")
    expect(box).to_have_attribute("aria-live", "polite")
    expect(box).not_to_have_attribute("aria-hidden")
    assert PEAK_CLAUSE.search(box.inner_text()), box.inner_text()
    assert box.inner_text().strip() == line.strip(), (box.inner_text(), line)

    # HONESTY: the megabytes on the page ARE the terminal payload's
    # ``rss_peak_mb`` — one forced reading, two surfaces.
    assert terminal["rss_peak_mb"] is not None, terminal
    assert _clause_number(line) == terminal["rss_peak_mb"], (line, terminal)

    # And the number survives the page: a reload re-attaches the terminal
    # (initSyncButton) and re-renders the SAME clause.
    page.reload()
    expect(page.locator("#sync-progress-text")).to_have_text(
        SUCCESS_LINE, timeout=SETTLE_TIMEOUT_MS
    )
    assert _clause_number(page.locator("#sync-progress-text").inner_text()) == (
        terminal["rss_peak_mb"]
    )

    # No error surface on a success settle (unchanged contract).
    expect(page.locator("#sync-error-banner")).to_be_hidden()
    assert page.locator(".sync-modal-backdrop").count() == 0


# ---------------------------------------------------------------------------
# 2. The payload: additive, typed, terminal-kept — and silent while running
# ---------------------------------------------------------------------------


def test_status_payload_reports_rss(page: Page, app_url: str, db_ready: None) -> None:
    """The additive pin end to end, through the browser's own authenticated
    request surface: while the run is in flight the payload already carries
    the pair (``int | null``) and the pre-phase keys are byte-for-byte
    unchanged, while the PAGE shows no clause at all; after the terminal the
    peak is a positive integer that stays with the run, and the pair keeps
    the invariant the page relies on (``peak >= current`` — the sampling
    point lifts the lazily-refreshed high-water, ``MB_SLACK`` is 0).
    """
    page.set_default_timeout(30_000)
    _start_from_page(page, app_url)

    # --- during `running` ------------------------------------------------
    running = _wait_sync_state(page, app_url, {"running"})
    assert running["state"] == "running", running
    # The additive contract on the wire: exactly the pre-phase keys plus
    # the memory pair — nothing renamed, nothing dropped.
    assert set(running) == PRE_187_STATUS_KEYS | MEMORY_KEYS, set(running)
    for key, value in running.items():
        if key in MEMORY_KEYS:
            # The documented type: int | null (null is "no reading").
            assert value is None or isinstance(value, int), (key, value)
        elif key == "detail":
            assert isinstance(value, dict), (key, value)
        elif key in _COUNT_KEYS:
            assert isinstance(value, int), (key, value)
        else:
            assert value is None or isinstance(value, (str, int, bool)), (key, value)
    # The types are ints on a platform that knows them (the run start
    # FORCES the first sample, so a running tick is never null here).
    if MEMORY_KNOWABLE:
        assert running["rss_mb"] is not None, running
        assert running["rss_peak_mb"] is not None, running
        assert running["rss_peak_mb"] >= running["rss_mb"] - MB_SLACK, running

    # The RUNNING state shows NO clause: the box carries the live label and
    # says nothing about memory (the clause is terminal-only in the UI). The
    # word is the tell (the fixture names never carry it), so a clause that
    # leaked into a running label fails even when its number is null-ish.
    live_text = page.locator("#sync-progress-text").inner_text()
    assert live_text.startswith("Syncing…"), live_text
    assert not _ANY_PEAK.search(live_text), live_text
    assert not _ANY_PEAK.search(page.locator("#sync-progress").inner_text())

    # --- after the terminal ---------------------------------------------
    terminal = _wait_sync_state(page, app_url, {"success", "failed", "cancelled"})
    assert terminal["state"] == "success", terminal
    assert set(terminal) == PRE_187_STATUS_KEYS | MEMORY_KEYS, set(terminal)
    current, peak = terminal["rss_mb"], terminal["rss_peak_mb"]
    if MEMORY_KNOWABLE:
        assert peak is not None and isinstance(peak, int), terminal
        assert peak > 0, terminal  # a real reading, never a placeholder 0
        assert current is not None and current > 0, terminal
        assert peak >= current - MB_SLACK, (current, peak)
    else:
        assert peak is None and current is None, terminal

    # The pair SURVIVES the run (no reset between the terminal and a later
    # poll — the page can be reloaded an hour later and still answers).
    again = _status(page, app_url)
    assert (again["rss_mb"], again["rss_peak_mb"]) == (current, peak), again
    # …and the pre-existing values rode the terminal unchanged too.
    assert again["state"] == "success"
    assert again["error"] is None
    assert isinstance(again["detail"], dict) and again["detail"]["added"] == N_FILES


# ---------------------------------------------------------------------------
# 3. Anonymous: no memory number on any anonymous surface (A4)
# ---------------------------------------------------------------------------


def test_anonymous_cannot_read_memory(
    page: Page, browser: Browser, app_url: str, db_ready: None
) -> None:
    """The exposure guard is the status route's own admin gate: a logged-out
    visitor gets the pre-phase 403 with the pre-phase body, before or after
    a run — and the Sources shell shows the sign-in gate with no sync button
    and no memory clause anywhere in the DOM.

    The probe runs in a SEPARATE browser context (``browser.new_context()``)
    — this test's own page is logged in by the run above, and
    ``page.context.new_page()`` would SHARE that session's cookies, which
    would make the "anonymous" probe an admin one."""
    # A run FIRST, so the guard is proven with real numbers in existence.
    _start_from_page(page, app_url)
    terminal = _wait_sync_state(page, app_url, {"success"})
    if MEMORY_KNOWABLE:
        assert terminal["rss_peak_mb"] is not None, terminal

    ctx = browser.new_context(viewport={"width": 1280, "height": 800})
    try:
        anon = ctx.new_page()
        r = anon.request.get(f"{app_url}/api/sync/status")
        assert r.status == 403, r.text
        assert r.json() == {"detail": "admin only"}

        # The anonymous shell: the gate is the anonymity proof, the sync
        # controls are never revealed, and nothing about memory renders.
        anon.goto(f"{app_url}/sources.html")
        expect(anon.locator("#sources-gate")).to_be_visible(timeout=30_000)
        expect(anon.locator("#sync-btn")).to_be_hidden()
        expect(anon.locator("#sync-progress")).to_be_hidden()
        assert not _ANY_PEAK.search(anon.locator("body").inner_text())
    finally:
        ctx.close()


# ---------------------------------------------------------------------------
# 4. The cancelled terminal keeps the peak (the test_sync_cancel shape)
# ---------------------------------------------------------------------------


def test_cancelled_terminal_keeps_the_peak(
    page: Page, app_url: str, db_ready: None
) -> None:
    """Cancel mid-run: the run lands in the NEUTRAL ``cancelled`` terminal
    (no banner, no modal, retry-ready) and the box still carries the
    clause — "Sync cancelled · peak <N> MB" — equal to the cancelled
    payload's ``rss_peak_mb``. A cancelled run's memory is still data: the
    phase-64 keep-final-values convention covers it like the counts."""
    page.set_default_timeout(30_000)
    _start_from_page(page, app_url)

    _wait_sync_state(page, app_url, {"running"})
    cancel = page.locator("#sync-cancel-btn")
    expect(cancel).to_be_visible()
    expect(cancel).to_be_enabled()
    cancel.click()
    expect(cancel).to_be_disabled()  # never stale, one request at a time

    body = _wait_sync_state(page, app_url, {"cancelled"}, timeout_s=30.0)
    assert body["state"] == "cancelled"
    assert body["error"] is None  # a cancel is intent, not failure
    assert body["detail"] == {}

    # The neutral settle, clause and all (the 2 s poll, one tick later).
    expect(page.locator("#sync-label")).to_have_text("Sync sources", timeout=30_000)
    expect(page.locator("#sync-btn")).to_be_enabled()
    expect(cancel).to_be_hidden()
    expect(page.locator("#sync-error-banner")).to_be_hidden()
    assert page.locator(".sync-modal-backdrop").count() == 0

    line = page.locator("#sync-progress-text").inner_text()
    if MEMORY_KNOWABLE:
        assert body["rss_peak_mb"] is not None, body
        assert body["rss_peak_mb"] > 0, body
        assert CANCELLED_LINE.fullmatch(line), line
        assert _clause_number(line) == body["rss_peak_mb"], (line, body)
    else:
        assert line == "Sync cancelled", line
    # The pair survives the cancel exactly like the counts do.
    again = _status(page, app_url)
    assert (again["rss_mb"], again["rss_peak_mb"]) == (
        body["rss_mb"],
        body["rss_peak_mb"],
    ), again


# ---------------------------------------------------------------------------
# 5. PLAN §7 at 375 px: the clause fits, the page never scrolls sideways
# ---------------------------------------------------------------------------


def test_peak_clause_fits_at_375(page: Page, app_url: str, db_ready: None) -> None:
    """The clause at phone width: the result line wraps INSIDE the bordered
    box (``overflow-wrap: anywhere``) — nothing clipped, no horizontal page
    overflow, the box inside the viewport (AGENTS rule 5 / PLAN §7). The
    clause is a few characters of trailing text, so a layout shift or a
    clipped tail at 375 px is exactly what this pins out."""
    _open_sources(page, app_url)
    page.set_viewport_size(MOBILE_VIEWPORT)  # the phone width, then the click
    page.wait_for_timeout(150)  # the reflow settles before the run paints
    _click_sync(page)
    terminal = _wait_sync_state(page, app_url, {"success"})

    text_el = page.locator("#sync-progress-text")
    expect(text_el).to_have_text(
        SUCCESS_LINE if MEMORY_KNOWABLE else f"{N_FILES} added",
        timeout=SETTLE_TIMEOUT_MS,
    )
    line = text_el.inner_text()
    if MEMORY_KNOWABLE:
        assert PEAK_CLAUSE.search(line), line
        assert _clause_number(line) == terminal["rss_peak_mb"], (line, terminal)

    # The box + its text do not clip, the box stays in the viewport, and
    # the PAGE itself never scrolls sideways at 375 px.
    page.wait_for_timeout(150)  # the reflow settles before the measure
    geometry = page.evaluate(
        """() => {
            const box = document.querySelector('#sync-progress');
            const textEl = document.querySelector('#sync-progress-text');
            const b = box.getBoundingClientRect();
            return {
                boxScrollWidth: box.scrollWidth,
                boxClientWidth: box.clientWidth,
                textScrollWidth: textEl.scrollWidth,
                textClientWidth: textEl.clientWidth,
                left: b.left,
                right: b.right,
                height: b.height,
                docScroll: document.documentElement.scrollWidth,
                innerWidth: window.innerWidth,
                text: textEl.textContent,
            };
        }"""
    )
    assert geometry["height"] > 0, geometry
    assert geometry["boxScrollWidth"] <= geometry["boxClientWidth"], geometry
    assert geometry["textScrollWidth"] <= geometry["textClientWidth"], geometry
    assert geometry["left"] >= 0, geometry
    assert geometry["right"] <= MOBILE_VIEWPORT["width"] + 1, geometry
    assert geometry["docScroll"] <= geometry["innerWidth"] + 1, geometry
    assert PEAK_CLAUSE.search(geometry["text"]) or not MEMORY_KNOWABLE, geometry