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

"""Phase 79 E2E (Playwright): API tokens — the owner's sentence, pinned
in a real browser.

TODO.md L5 (owner 2026-09-06): "Add api tokens that the admin can
generate and hand out so people can log in to use the app. The only
thing that should be accessible without an API token is shared chats.
The web ui should ask for a token before letting a user through and
should cache that token in browser storage so they don't have to keep
entering it."

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

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

Test → claim mapping (every clause of the owner's sentence is pinned):

1. ``test_anonymous_is_locked_out`` — "ask for a token before letting a
   user through": a fresh context meets ``#auth-gate`` on the chat
   page, ``#main`` is inert (the composer is NOT keyboard-reachable —
   the inverted tab-order walk), and the three gated endpoints
   (chat / suggestions / document content) 401 from the context's own
   empty cookies.
2. ``test_shared_chats_stay_open_anonymous`` — "the only thing that
   should be accessible without an API token is shared chats": the
   admin creates + shares a saved chat (the house API pattern), a
   FRESH context opens ``/shared/<token>`` anonymously and sees the
   conversation rendered — no gate anywhere on that page.
3. ``test_admin_generates_token_in_ui`` — "the admin can generate":
   the admin navigates to the Tokens view, labels a token
   "e2e-alice" and generates — ``#token-once-value`` carries
   ``bor_`` + 32 hex (the A4 plaintext-once), the table shows the
   Active row, and the once-block is GONE on a re-show (the plaintext
   can never be re-shown).
4. ``test_token_user_uses_the_app`` — "hand out so people can log in
   and use the app": a fresh context signs in through the real gate
   (the task-04 ``login_with_token`` helper), chats end-to-end
   (mock LLM), opens a cited document in the same-page modal, and
   gets the role-``user`` header contract — every admin nav link
   absent, Sign out visible.
5. ``test_cached_token_survives_reload`` — "cache that token in
   browser storage so they don't have to keep entering it": the
   entered token lands in ``localStorage["bor.token"]``; a reload
   re-auths silently — no gate, no re-entry, still role user.
6. ``test_admin_only_walls_403_for_token_user`` — "every existing
   admin-only surface stays admin-only": the token user's own session
   cookie 403s on tokens / chats / docs / steering / git-sources.
7. ``test_sign_out_clears_the_cached_token`` — sign out clears the
   session AND the cached token (one logout, both gone); the gate
   comes back.
8. ``test_revocation_closes_the_door`` — "revocation is enforced
   IMMEDIATELY": the admin revokes through the UI two-step; the
   holder's next gated request 401s (the 401 clears the dead
   session cookie — the next whoami is anonymous), and a FRESH login
   attempt with the same token is refused at the gate.
9. ``test_wrong_token_is_one_generic_error`` — no enumeration: a
   wrong token shows the gate's role=alert line and keeps the
   visitor anonymous; the API's error body for a malformed token is
   byte-equal to the one for a well-formed unknown token.

DB isolation: the shared e2e Postgres keeps ``api_tokens`` (and
``saved_chats``) rows across suites. This file is the only suite that
issues tokens, so an autouse fixture deletes the ``e2e-``-labeled
rows before each test (never a TRUNCATE — the shared DB may hold the
owner's real tokens); the shared-chat test deletes its own saved row
in a ``finally``. Every scenario runs in its OWN fresh browser
context — no cached token (localStorage) or session cookie leaks
between tests.
"""
from __future__ import annotations

import asyncio
import re
from pathlib import Path
from threading import Thread
from typing import Any

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

from app.config import Settings
from app.db import SessionLocal
from app.rag.importer import ImportSummary, import_sources
from app.rag.llm import LLMClient
from e2e.auth_helpers import login, login_with_token

REPO = Path(__file__).resolve().parents[2]
FIXTURES = REPO / "tests" / "fixtures" / "docs"
MOCK_ANSWER_MARKER = "Deterministic mock answer for E2E"
#: The plaintext token's shape (owner-locked A4): prefix + 32 hex.
TOKEN_RE = re.compile(r"bor_[0-9a-f]{32}")
#: The five admin-only nav links (the role-`user` contract: ALL of
#: them stay absent for a token user — header.js reveals them only
#: for role === "admin").
ADMIN_NAV_LINKS = ("#nav-sources", "#nav-git-sources", "#nav-tuning", "#nav-history", "#nav-tokens")


async def _import_fixtures(mock_port: int) -> ImportSummary:
    kwargs: dict[str, Any] = {"_env_file": None, "llm_base_url": f"http://127.0.0.1:{mock_port}/v1"}
    settings = Settings(**kwargs)  # pyright: ignore[reportCallIssue]
    return await import_sources([FIXTURES], LLMClient(settings))


def _run_in_thread(coro: Any) -> Any:
    """Run a coroutine on a worker thread.

    Playwright's sync API keeps an asyncio loop running on the test
    thread, so ``asyncio.run`` cannot be called directly from a test
    body.
    """
    box: dict[str, Any] = {}

    def runner() -> None:
        try:
            box["value"] = asyncio.run(coro)
        except BaseException as e:  # noqa: BLE001 — re-raised on the test thread
            box["error"] = e

    t = Thread(target=runner)
    t.start()
    t.join()
    if "error" in box:
        raise box["error"]
    return box["value"]


def _reset_db(mock_port: int, seed: bool) -> ImportSummary | None:
    """Truncate the KB (and query log + steering notes — deterministic
    mock answers), then optionally re-import fixtures. ``saved_chats``
    and ``api_tokens`` are deliberately NOT touched (the house
    pattern)."""
    with SessionLocal() as db:
        db.execute(text("TRUNCATE chunks, documents, query_log, steering_notes"))
        db.commit()
    if not seed:
        return None
    return _run_in_thread(_import_fixtures(mock_port))


def _cleanup_e2e_tokens() -> None:
    """Delete this suite's issued tokens (deterministic re-runs).

    Label-scoped on ``e2e-`` — never a TRUNCATE: the shared e2e DB is
    also the dev DB and may hold the owner's real tokens.
    """
    with SessionLocal() as db:
        db.execute(text("DELETE FROM api_tokens WHERE label LIKE 'e2e-%'"))
        db.commit()


@pytest.fixture(autouse=True)
def _e2e_tokens_clean(db_ready: None) -> None:
    """Start every test from the same token-empty state (this file is
    the only E2E suite that issues tokens)."""
    _cleanup_e2e_tokens()


def _ask(page: Page, question: str) -> None:
    """Send one turn and wait until the grounded answer has fully
    landed (the ``done`` event restored the Send button)."""
    page.fill("#message-input", question)
    page.click("#send-btn")
    expect(page.locator(".msg.user .bubble").last).to_contain_text(question)
    expect(page.locator(".msg.brain .bubble").last).to_contain_text(
        MOCK_ANSWER_MARKER, timeout=30_000
    )
    expect(page.locator("#send-btn")).to_be_enabled()
    expect(page.locator("#send-label")).to_have_text("Send")


def _cookies(page: Page) -> dict[str, str]:
    """The session cookies the browser context holds (the test's API
    side sees exactly what that browser sees)."""
    return {
        c["name"]: c["value"]
        for c in page.context.cookies()
        if "name" in c and "value" in c
    }


def _create_token(app_url: str, cookies: dict[str, str], label: str) -> tuple[str, str]:
    """Admin-issued token through the house API pattern (httpx + the
    signed admin cookie): ``POST /api/tokens`` → 201 (the ONE response
    that carries the plaintext, A4). Returns (row id, plaintext)."""
    r = httpx.post(f"{app_url}/api/tokens", json={"label": label}, cookies=cookies, timeout=10)
    assert r.status_code == 201, r.text
    body = r.json()
    assert body["label"] == label
    assert TOKEN_RE.fullmatch(body["token"]), f"bad token shape: {body}"
    return body["id"], body["token"]


def _whoami(page: Page) -> dict[str, Any]:
    """The page's own ``/api/whoami`` read (the context's cookies)."""
    return page.evaluate("() => fetch('/api/whoami').then((r) => r.json())")


# ---------------------------------------------------------------------------
# 1. Anonymous is locked out: the gate is up, #main is inert, and the
#    three app surfaces 401 (only the shared chats are open — test 2)
# ---------------------------------------------------------------------------


def test_anonymous_is_locked_out(page: Page, app_url: str, db_ready: None) -> None:
    page.set_default_timeout(30_000)
    js_errors: list[str] = []
    page.on("pageerror", lambda e: js_errors.append(str(e)))

    page.goto(app_url)

    # The gate is the visible surface of an anonymous chat page…
    gate = page.locator("#auth-gate")
    expect(gate).to_be_visible(timeout=30_000)
    expect(gate).to_have_attribute("aria-labelledby", "auth-gate-title")
    expect(page.locator("#auth-gate-input")).to_be_focused()  # the gate takes the focus
    expect(page.locator("#auth-gate-form button[type=submit]")).to_be_visible()
    expect(page.locator("#auth-gate-error")).to_have_attribute("role", "alert")

    # …and it LOCKS the app: #main is inert while the gate is up, so
    # the composer cannot be reached — not by mouse, not by keyboard.
    assert page.evaluate("() => document.getElementById('main').inert === true")

    # Inverted tab-order walk (the test_suggestion_chips keyboard
    # walk, inverted): from the page start, Tab cycles the gate and
    # the sign-in link only — the composer is never a tab stop.
    seen: list[str] = []
    for _ in range(12):
        page.keyboard.press("Tab")
        seen.append(
            page.evaluate(
                "() => (document.activeElement && document.activeElement.id) || ''"
            )
        )
    assert "auth-gate-input" in seen, f"the gate input must be keyboard-reachable: {seen}"
    assert "message-input" not in seen, (
        f"the composer must NOT be keyboard-reachable while the gate is up: {seen}"
    )

    # The API agrees, from the context's own (empty) cookies: the
    # three app surfaces all refuse with ONE 401 detail.
    anon_chat = page.evaluate(
        """async () => {
            const r = await fetch('/api/chat', {
                method: 'POST',
                headers: {'Content-Type': 'application/json'},
                body: JSON.stringify({message: 'hello?'}),
            });
            return {status: r.status, body: await r.json()};
        }"""
    )
    assert anon_chat["status"] == 401, anon_chat
    assert anon_chat["body"] == {"detail": "authentication required"}, anon_chat

    anon_sugg = page.evaluate(
        """() => fetch('/api/suggestions')
            .then((r) => r.json().then((body) => ({status: r.status, body})))"""
    )
    assert anon_sugg["status"] == 401, anon_sugg
    assert anon_sugg["body"] == {"detail": "authentication required"}, anon_sugg

    anon_doc = page.evaluate(
        """async () => {
            const r = await fetch(
                '/api/documents/content?source=docs&path=homelab%2Fkubernetes.md');
            return {status: r.status, body: await r.json()};
        }"""
    )
    assert anon_doc["status"] == 401, anon_doc
    assert anon_doc["body"] == {"detail": "authentication required"}, anon_doc

    # The gated boot itself must be crash-free.
    assert not js_errors, f"the gated boot must not throw: {js_errors}"


# ---------------------------------------------------------------------------
# 2. Shared chats stay open: the ONLY anonymous content (the owner's
#    sentence) — a fresh context reads the shared conversation with no
#    gate anywhere
# ---------------------------------------------------------------------------


def test_shared_chats_stay_open_anonymous(
    page: Page, browser: Browser, app_url: str, db_ready: None
) -> None:
    page.set_default_timeout(30_000)
    login(page, app_url, next="/")

    # As admin: create + share a saved chat (the house API pattern
    # from test_share_chat — httpx with the signed session cookie).
    q = "How is my Kubernetes cluster set up? (api-tokens-shared)"
    r = httpx.post(
        f"{app_url}/api/chats",
        json={
            "messages": [
                {"who": "user", "text": q},
                {"who": "brain", "text": "Deterministic mock answer for E2E (api-tokens-shared)"},
            ]
        },
        cookies=_cookies(page),
        timeout=10,
    )
    assert r.status_code == 201, r.text
    chat_id: str = r.json()["id"]
    r = httpx.post(f"{app_url}/api/chats/{chat_id}/share", cookies=_cookies(page), timeout=10)
    assert r.status_code == 200, r.text
    share_url: str = r.json()["share_url"]
    assert share_url.startswith("/shared/"), share_url
    title = " ".join(q.split())[:120]  # the auto-title convention

    anon_ctx: BrowserContext | None = None
    try:
        # The anonymous JSON snapshot is open (no session at all)…
        snap = httpx.get(f"{app_url}/api{share_url}", timeout=10)
        assert snap.status_code == 200, snap.text
        assert snap.json()["title"] == title

        # …and the shared PAGE renders the conversation in a FRESH
        # context (no cookies, no cached token) — with no gate
        # anywhere on that page: shared chats are the anonymous
        # surface, full stop.
        anon_ctx = browser.new_context()
        anon = anon_ctx.new_page()
        anon.set_default_timeout(30_000)
        anon.goto(app_url + share_url)
        expect(anon.locator("#shared-title")).to_have_text(title)
        expect(anon.locator(".msg.user .bubble")).to_contain_text(q)
        expect(anon.locator(".msg.brain .bubble")).to_contain_text("api-tokens-shared")
        expect(anon.locator(".auth-gate")).to_have_count(0)
        expect(anon.locator("#auth-gate, #doc-auth-gate")).to_have_count(0)
        # The guest header offers sign-in (the shared page contract).
        expect(anon.locator("#sign-in-link")).to_be_visible(timeout=15_000)
    finally:
        if anon_ctx is not None:
            anon_ctx.close()
        httpx.delete(f"{app_url}/api/chats/{chat_id}", cookies=_cookies(page), timeout=10)


# ---------------------------------------------------------------------------
# 3. The admin generates a token in the UI: the plaintext appears
#    EXACTLY ONCE and the Active row lands in the table; the
#    once-block is gone on a re-show
# ---------------------------------------------------------------------------


def test_admin_generates_token_in_ui(page: Page, app_url: str, db_ready: None) -> None:
    page.set_default_timeout(30_000)
    login(page, app_url, next="/")
    expect(page).to_have_url(app_url + "/", timeout=30_000)

    # The Tokens view is the shell's sixth nav link (the phase-76
    # fold) — revealed for the admin, hidden for everyone else.
    expect(page.locator("#nav-tokens")).to_be_visible(timeout=15_000)
    page.click("#nav-tokens")
    expect(page.locator("#view-tokens")).to_be_visible()
    expect(page.locator("#tokens-gate")).to_be_hidden()  # admin: no sign-in gate
    expect(page.locator("#token-create")).to_be_visible(timeout=15_000)

    # Generate: label "e2e-alice" → the plaintext appears exactly
    # once, in the mono read-only field (the A4 shown-once contract).
    page.fill("#token-label", "e2e-alice")
    page.click("#token-generate")
    expect(page.locator("#token-once")).to_be_visible(timeout=15_000)
    token = page.input_value("#token-once-value")
    assert TOKEN_RE.fullmatch(token), f"bad token shape: {token!r}"
    expect(page.locator("#tokens-status")).to_have_text(
        "Token created — copy it now; it won't be shown again."
    )

    # The table shows the Active row (phase 101 D1: the Status column
    # is gone — the four-column active table IS the status): no
    # Revoked pill, a Revoke action.
    row = page.locator("#tokens-tbody tr", has_text="e2e-alice")
    expect(row).to_have_count(1)
    expect(row.locator("td.tokens-label-cell")).to_have_text("e2e-alice")
    expect(row.locator(".stale-pill")).to_have_count(0)
    expect(row.locator("button.token-revoke")).to_have_count(1)

    # The once-block is NOT re-shown on a re-show: nav away (RAG) and
    # back — the router's re-show refresh re-runs the list load,
    # which hides + wipes the once-block. The plaintext is gone.
    page.click("#nav-sources")
    expect(page.locator("#view-rag")).to_be_visible()
    page.go_back()
    expect(page.locator("#view-tokens")).to_be_visible(timeout=15_000)
    expect(page.locator("#token-once")).to_be_hidden()
    assert page.input_value("#token-once-value") == "", "the plaintext must be wiped on re-show"
    # The row survives the re-render (the token itself is unaffected).
    expect(page.locator("#tokens-tbody tr", has_text="e2e-alice")).to_have_count(1)


# ---------------------------------------------------------------------------
# 4. The token flow: a fresh context signs in through the real gate
#    and USES the app — a grounded chat turn (mock LLM; phase 119,
#    LOCKED A1: the turn is the scripted summary-read flow — chips
#    cite READ docs only, and this turn's read earns the cited chip),
#    a cited document opened in the same-page modal, the role-user
#    header
# ---------------------------------------------------------------------------


def test_token_user_uses_the_app(
    page: Page, browser: Browser, app_url: str, mock_llm: int, db_ready: None
) -> None:
    page.set_default_timeout(30_000)
    _reset_db(mock_llm, seed=True)  # a deterministic seeded KB for the grounded turn

    # The admin generates + hands out the token (fresh context per
    # role — the admin's browser is never the holder's browser).
    login(page, app_url, next="/")
    _id, token = _create_token(app_url, _cookies(page), "e2e-bob")

    user_ctx: BrowserContext | None = None
    try:
        user_ctx = browser.new_context()
        user = user_ctx.new_page()
        user.set_default_timeout(30_000)

        # The holder signs in through the REAL in-app gate (the
        # task-04 helper: fill #auth-gate-input → submit → gate hides).
        login_with_token(user, app_url, token)

        # Use the app: a GROUNDED turn against the seeded KB (mock
        # LLM). Phase 119 (LOCKED A1): chips cite READ docs only — a
        # plain question would chip nothing (the retired phase-118 A4
        # union is gone), so the turn is the mock's scripted
        # summary-read flow (SUMMARY_SEED_READ_TRIGGER): it reads the
        # kubernetes fixture, and that read is the chip's source.
        q = (
            "Read the suggested document: read docs/homelab/kubernetes.md — "
            "how is my Kubernetes cluster set up? (api-tokens-flow)"
        )
        user.fill("#message-input", q)
        user.click("#send-btn")
        expect(user.locator(".msg.user .bubble").last).to_contain_text(q)

        # The turn's cited source chip (its one READ doc — the scripted
        # read, phase 119 A1) opens the document in the SAME-PAGE modal
        # (the require_user content endpoint passes for a live token
        # session).
        chip = user.locator(".msg.brain a.source-chip").first
        expect(chip).to_be_visible(timeout=30_000)
        expect(user.locator("#send-label")).to_have_text("Send")
        chip.click()
        expect(user.locator("#doc-modal")).to_be_visible()
        expect(
            user.locator("#doc-modal-content .doc-md, #doc-modal-content pre.doc-raw")
        ).to_have_count(1, timeout=15_000)  # the document rendered (not the loading line)
        expect(user.locator("#doc-modal-content .doc-modal-loading")).to_have_count(0)
        expect(user.locator("#doc-modal-title")).not_to_be_empty()
        expect(user.locator("#doc-modal-title")).not_to_have_text("Loading…")
        expect(user.locator("#doc-modal-title")).not_to_have_text("Document not found")

        # The role-`user` header contract: ALL five admin nav links
        # are absent (they reveal only for role === "admin"), and the
        # auth pair is the signed-in branch (Sign out, no Sign in).
        for link in ADMIN_NAV_LINKS:
            expect(user.locator(link)).to_be_hidden()
        expect(user.locator("#sign-in-link")).to_be_hidden()
        expect(user.locator("#sign-out-btn")).to_be_visible()

        # The server agrees: authenticated, role user — NOT admin.
        who = _whoami(user)
        assert who == {"authenticated": True, "role": "user"}
    finally:
        if user_ctx is not None:
            user_ctx.close()


# ---------------------------------------------------------------------------
# 5. Caching: the token lands in localStorage and a reload re-auths
#    silently — no gate, no re-entry
# ---------------------------------------------------------------------------


def test_cached_token_survives_reload(
    page: Page, browser: Browser, app_url: str, db_ready: None
) -> None:
    page.set_default_timeout(30_000)
    login(page, app_url, next="/")
    _id, token = _create_token(app_url, _cookies(page), "e2e-cache")

    user_ctx: BrowserContext | None = None
    try:
        user_ctx = browser.new_context()
        user = user_ctx.new_page()
        user.set_default_timeout(30_000)

        # The gate's success path caches the entered token in
        # localStorage (the owner's sentence: "cache that token in
        # browser storage").
        login_with_token(user, app_url, token)
        assert user.evaluate("() => localStorage.getItem('bor.token')") == token

        # A reload re-auths SILENTLY from the cache: no gate, no
        # re-entry — the chat UI is interactive straight away, the
        # lock is released, the role is still user.
        user.reload()
        expect(user.locator("#auth-gate")).to_be_hidden(timeout=30_000)
        expect(user.locator("#message-input")).to_be_visible()
        assert user.evaluate("() => document.getElementById('main').inert === false")
        assert _whoami(user) == {"authenticated": True, "role": "user"}
        # The cache is what re-authed the page — it survived the reload.
        assert user.evaluate("() => localStorage.getItem('bor.token')") == token
    finally:
        if user_ctx is not None:
            user_ctx.close()


# ---------------------------------------------------------------------------
# 6. The admin-only walls: every admin surface 403s the token user's
#    own session cookie (require_admin, unchanged)
# ---------------------------------------------------------------------------


def test_admin_only_walls_403_for_token_user(
    page: Page, browser: Browser, app_url: str, db_ready: None
) -> None:
    page.set_default_timeout(30_000)
    login(page, app_url, next="/")
    _id, token = _create_token(app_url, _cookies(page), "e2e-wall")

    user_ctx: BrowserContext | None = None
    try:
        user_ctx = browser.new_context()
        user = user_ctx.new_page()
        user.set_default_timeout(30_000)
        login_with_token(user, app_url, token)
        cookies = _cookies(user)

        # The token user's OWN signed session cookie 403s on every
        # admin surface — "admin only" (the phase-16 contract),
        # never a 401 (the user IS authenticated — just not an
        # admin).
        for method, url in (
            ("GET", "/api/tokens"),
            ("GET", "/api/chats"),
            ("GET", "/api/docs"),
            ("POST", "/api/steering"),
            ("GET", "/api/git-sources"),
        ):
            r = httpx.request(
                method,
                app_url + url,
                cookies=cookies,
                timeout=10,
                json={"note": "wall check"} if method == "POST" else None,
            )
            assert r.status_code == 403, (method, url, r.status_code, r.text)
            assert r.json() == {"detail": "admin only"}, (method, url, r.text)
    finally:
        if user_ctx is not None:
            user_ctx.close()


# ---------------------------------------------------------------------------
# 7. Sign out: the header binding clears the session AND the cached
#    token — the gate comes back
# ---------------------------------------------------------------------------


def test_sign_out_clears_the_cached_token(
    page: Page, browser: Browser, app_url: str, db_ready: None
) -> None:
    page.set_default_timeout(30_000)
    login(page, app_url, next="/")
    _id, token = _create_token(app_url, _cookies(page), "e2e-signout")

    user_ctx: BrowserContext | None = None
    try:
        user_ctx = browser.new_context()
        user = user_ctx.new_page()
        user.set_default_timeout(30_000)
        login_with_token(user, app_url, token)
        assert user.evaluate("() => localStorage.getItem('bor.token')") == token

        # Sign out (the header binding): POST /api/logout + drop the
        # cached token + reload — ONE logout clears both the server
        # session and the localStorage key.
        user.click("#sign-out-btn")
        expect(user.locator("#auth-gate")).to_be_visible(timeout=30_000)
        assert user.evaluate("() => localStorage.getItem('bor.token')") is None
        assert _whoami(user) == {"authenticated": False, "role": "anonymous"}
    finally:
        if user_ctx is not None:
            user_ctx.close()


# ---------------------------------------------------------------------------
# 8. Revocation closes the door: the admin's UI two-step kills the
#    token immediately — the holder's next request 401s (and the
#    401 clears the dead session cookie), and a fresh login attempt
#    with the same token is refused
# ---------------------------------------------------------------------------


def test_revocation_closes_the_door(
    page: Page, browser: Browser, app_url: str, db_ready: None
) -> None:
    page.set_default_timeout(30_000)
    login(page, app_url, next="/")
    _id, token = _create_token(app_url, _cookies(page), "e2e-revoke")

    user_ctx: BrowserContext | None = None
    fresh_ctx: BrowserContext | None = None
    try:
        # The holder is signed in (a fresh context of their own).
        user_ctx = browser.new_context()
        user = user_ctx.new_page()
        user.set_default_timeout(30_000)
        login_with_token(user, app_url, token)

        # The admin revokes through the UI two-step (Revoke → Yes —
        # the inline confirm, no native dialog).
        page.goto(app_url + "/tokens.html")
        row = page.locator("#tokens-tbody tr", has_text="e2e-revoke")
        expect(row).to_have_count(1, timeout=15_000)
        row.locator("button.token-revoke").click()
        expect(row.locator(".history-confirm-yes")).to_be_visible()
        row.locator(".history-confirm-yes").click()
        # Phase 101 (D1): the row LEAVES the active table and lands in
        # the revoked table below with its Revoked date cell (the
        # em-dash / .stale-pill column is gone from both tables — the
        # table's position IS the status).
        revoked = page.locator("#tokens-revoked-tbody tr", has_text="e2e-revoke")
        expect(revoked).to_have_count(1, timeout=15_000)
        expect(revoked.locator(".stale-pill")).to_have_count(0)
        expect(revoked.locator("td.tokens-date-cell").last).not_to_be_empty()
        expect(page.locator("#tokens-status")).to_have_text('Revoked "e2e-revoke".')

        # Enforcement is IMMEDIATE on the holder's next request (the
        # session stores the row id — no server-side session store;
        # the live row check IS the revocation check). Driven with the
        # USER context's own cookies (its real browser cookie jar —
        # response headers land exactly as in a live browser):
        #
        #   1. whoami is LAZY by contract — the dead session still
        #      reports "user" (the endpoint does not live-check)…
        assert _whoami(user) == {"authenticated": True, "role": "user"}
        #   2. …but the next GATED request 401s — and the 401's
        #      Set-Cookie clears the dead session from the jar…
        resp = user.evaluate(
            """async () => {
                const r = await fetch('/api/chat', {
                    method: 'POST',
                    headers: {'Content-Type': 'application/json'},
                    body: JSON.stringify({message: 'hello?'}),
                });
                return {status: r.status, body: await r.json()};
            }"""
        )
        assert resp["status"] == 401, resp
        assert resp["body"] == {"detail": "authentication required"}, resp
        #   3. …so from now on the holder is genuinely anonymous.
        assert _whoami(user) == {"authenticated": False, "role": "anonymous"}

        # A FRESH login attempt with the same (now revoked) token is
        # refused at the gate: the error line shows, the gate stays,
        # the visitor is still anonymous.
        fresh_ctx = browser.new_context()
        fresh = fresh_ctx.new_page()
        fresh.set_default_timeout(30_000)
        login_with_token(fresh, app_url, token, expect_error=True)
        assert _whoami(fresh) == {"authenticated": False, "role": "anonymous"}
    finally:
        if user_ctx is not None:
            user_ctx.close()
        if fresh_ctx is not None:
            fresh_ctx.close()


# ---------------------------------------------------------------------------
# 9. Wrong token: the gate's one generic error, no enumeration — the
#    malformed and the well-formed-unknown failures are byte-equal
# ---------------------------------------------------------------------------


def test_wrong_token_is_one_generic_error(page: Page, app_url: str, db_ready: None) -> None:
    page.set_default_timeout(30_000)
    # The task-04 helper's default sentinel drives the wrong-token
    # contract (bor_ + 32 zeros — never a live token): the role=alert
    # error line shows, the gate stays, the visitor is anonymous.
    login_with_token(page, app_url)  # token defaults to the all-zeros sentinel
    assert _whoami(page) == {"authenticated": False, "role": "anonymous"}

    # No enumeration: the API's ONE generic 401 — the error body for
    # a wrong-FORMAT token equals the one for a well-formed but
    # unknown token (no shape hint on a credential endpoint).
    malformed = httpx.post(
        f"{app_url}/api/token-auth", json={"token": "not-a-token-at-all"}, timeout=10
    )
    unknown = httpx.post(
        f"{app_url}/api/token-auth", json={"token": "bor_" + "0" * 32}, timeout=10
    )
    assert malformed.status_code == 401
    assert unknown.status_code == 401
    assert malformed.json() == unknown.json() == {"detail": "invalid token"}