"""Phase 189 E2E (Playwright): a LONG agent turn saves — and Share publishes it.

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

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

WHY THESE TESTS EXIST (owner incident 2026-10-16, read this before editing)
--------------------------------------------------------------------------
Shared chat ``0101d2ca-b24c-4d13-9f65-c0837e993c50`` (row
``8b345ab5-aab3-4412-9642-0f869b3cec4d``) shows ONLY the user's question
although the LLM answered it completely. The app log for that row carries the
whole sequence:

* the per-turn line ``thinking=xhigh thinking_chars=23937 tool_calls=33
  rounds=34 cap_hit=0 total_ms=400173`` + ``POST /api/chat 200 stream=1`` —
  the client consumed the FULL 400 s stream, so the ``done`` frame landed;
* the done frame carried **22 sources** (the cited tier is the agent's READ
  docs since phase 119 — ``merge_cited_docs`` — not the retrieval top-N, and
  production runs ``BOR_AGENT_MAX_ROUNDS=50``), 33 tool calls and one
  thinking entry per contiguous thinking run;
* immediately after it: ``PUT /api/chats/8b345ab5… 422 Unprocessable
  Entity`` — the done-time auto-save was REJECTED at the phase-83 list caps
  sized for the round-cap-10 era (``sources`` ≤ 20, ``tools`` ≤ 50,
  ``segments`` ≤ 50);
* the phase-55 A2 quiet contract swallowed the failure (a one-line
  ``#send-status`` note, no banner) and NOTHING retried, so the row kept its
  pre-turn state — ``created_at == updated_at``, ``n_msgs = 1``, a
  single-write user-only row;
* ``POST …/share`` then minted its token on that stale row (84 other
  single-write user-only rows exist, 2 of them shared).

The four tests below pin the owner-visible contract of the phase's four
fixes, through the real UI and the real wire:

* **A1 (task 01) — the caps follow the real configuration.** The incident
  payload shape (22 ``sources`` / 33 ``tools`` / 51 ``segments``) is
  **201** on ``POST /api/chats`` and **200** on ``PUT /api/chats/<id>``,
  anonymously, and the row round-trips all three lists (pre-phase: both
  **422** — the incident). ``tests/integration/test_chats_api.py`` pins the
  over-cap variants at the new boundaries (201/201/451/101 → 422).
* **A3 (task 03) — a save point dropped by the in-flight guard re-runs.**
  A real mock turn's done-time auto-save lands even when the send-time
  auto-save is still parked at the server: the row ends with BOTH messages
  (pre-phase: the dropped save never came back and the row held 1).
* **A4 (task 03) — a failed save stays visible.** The one-line
  "Couldn't save automatically — will try on the next message." note is on
  screen after the failed auto-saves (no error banner — the A2 shape is
  unchanged) and the next successful save clears it.
* **A2 (task 02) — Share saves first.** With the auto-saves failing, the
  Share click UPSERTS the current conversation before minting the token
  (pinned in the request log: an in-cap ``PUT`` immediately precedes the
  ``POST …/share``), and the shared page renders the BRAIN answer — not just
  the question (pre-phase: exactly that, the frozen truncated snapshot).

Wire-shape notes for whoever edits the interception helpers: a Playwright
route handler that ``time.sleep``s blocks the sync dispatcher thread, so the
test thread cannot act during the hold — the delay-based "click while the
auto-save is parked" shape therefore races the parked (stale) write against
the share's own write for the row. Scenario 4 makes the row stale the
deterministic way instead (an injected **422** on the auto-saves — exactly
the incident's rejection, and one phase 189 A4 deliberately never retries),
which leaves no competing write in flight; scenario 3 keeps the hold and pins
the overlap PROPERLY by timestamping the answer's arrival inside the page (a
``MutationObserver``) and comparing it with the release time of the parked
request — so it cannot quietly degrade into a non-overlapping turn that passes
for the wrong reason. Scenario 3 holds the first WRITE (the ``POST`` create —
a fresh conversation's send-time save point creates the row; a ``PUT`` only
exists once a row is linked) and scenario 4's stale-row turn is a second turn
on a linked row, where the save points are ``PUT``s.

DB isolation: the shared e2e Postgres keeps ``saved_chats`` rows across
suites, so every test here uses a DISTINCTIVE question text (its auto-title
is therefore unique), never asserts on absolute row counts, and deletes the
rows it creates in a ``finally`` (admin cookie). The KB tables are truncated
+ re-seeded the house way (deterministic mock embeddings) ONCE per module —
these tests only need a GROUNDED mock answer, never a corpus change.
"""
from __future__ import annotations

import asyncio
import json
import re
import time
from collections.abc import Callable
from pathlib import Path
from threading import Thread
from typing import Any

import httpx
import pytest
from playwright.sync_api import Browser, BrowserContext, Page, Route, 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

REPO = Path(__file__).resolve().parents[2]
FIXTURES = REPO / "tests" / "fixtures" / "docs"
MOCK_ANSWER_MARKER = "Deterministic mock answer for E2E"
#: The phase-55 A2 quiet note (unchanged by phase 189 — only its LIFETIME
#: is longer: it now stays until the next successful save).
SAVE_NOTE = "Couldn't save automatically — will try on the next message."

#: The saved-chat WRITE surface as a URL pattern: the collection (the
#: ``POST /api/chats`` create) and one row (the ``PUT /api/chats/<id>``
#: update). The share mint (``POST /api/chats/<id>/share``) and the chat
#: SSE stream (``/api/chat``) deliberately do NOT match.
CHAT_WRITE_URL = re.compile(r"https?://[^/]+/api/chats(/[0-9a-f-]+)?$")
#: The share mint, observed (never held) so the request LOG proves the
#: upsert-before-token ordering of phase 189 A2.
CHAT_SHARE_URL = re.compile(r"https?://[^/]+/api/chats/[0-9a-f-]+/share$")

#: Long enough that the mock turn (≈ 1–2 s on this corpus) settles while the
#: first auto-save is still parked; short enough to keep the suite quick.
HOLD_S = 6.0

#: The page-side watch that stamps WHEN the answer text first reached the
#: DOM. Needed because the route handler's hold blocks the sync dispatcher
#: (see the module docstring): without a page-side clock the overlap
#: scenario could not prove the done-time save point arrived mid-upsert.
_WATCH_ANSWER_JS = """
(marker) => {
  window.__borAnswerAt = null;
  const check = () => {
    const bubbles = document.querySelectorAll('.msg.brain .bubble');
    if (!bubbles.length) return;
    const last = bubbles[bubbles.length - 1];
    if ((last.textContent || '').includes(marker)
        && window.__borAnswerAt === null) {
      window.__borAnswerAt = Date.now();
    }
  };
  new MutationObserver(check).observe(document.body, {
    childList: true, subtree: true, characterData: true,
  });
  check();
}
"""


# ---------------------------------------------------------------------------
# DB / app plumbing (the house pattern from test_chat_history.py)
# ---------------------------------------------------------------------------


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 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"]


@pytest.fixture(scope="module")
def kb(mock_llm: int) -> None:
    """Truncate the KB (and the query log — deterministic mock answers) and
    re-import the fixtures ONCE for the module: every scenario here only
    needs a grounded (non-deflected) mock answer. ``saved_chats`` is
    deliberately NOT touched — rows persist across suites and each test
    cleans up after itself."""
    with SessionLocal() as db:
        db.execute(text("TRUNCATE chunks, documents, query_log, steering_notes"))
        db.commit()
    _run_in_thread(_import_fixtures(mock_llm))


# ---------------------------------------------------------------------------
# Admin-side helpers (the API is the proof — auto-saves are SILENT, A2)
# ---------------------------------------------------------------------------


def _admin_cookies(page: Page) -> dict[str, str]:
    """The signed session cookies the browser holds after a form login."""
    return {
        c["name"]: c["value"]
        for c in page.context.cookies()
        if "name" in c and "value" in c
    }


def _auto_title(question: str) -> str:
    """The phase-50 auto-title convention: the first question,
    whitespace-collapsed, capped at 120 chars."""
    return " ".join(question.split())[:120]


def _find_row(
    app_url: str, cookies: dict[str, str], title: str
) -> dict[str, Any] | None:
    r = httpx.get(f"{app_url}/api/chats", timeout=10, cookies=cookies)
    assert r.status_code == 200
    return next((c for c in r.json()["chats"] if c["title"] == title), None)


def _row(app_url: str, cookies: dict[str, str], chat_id: str) -> dict[str, Any]:
    r = httpx.get(f"{app_url}/api/chats/{chat_id}", timeout=10, cookies=cookies)
    assert r.status_code == 200, r.text
    return r.json()


def _wait_row_messages(
    app_url: str,
    cookies: dict[str, str],
    title: str,
    messages: int,
    page: Page,
    timeout_s: float = 20.0,
) -> dict[str, Any]:
    """Poll the admin list until the row with this conversation's auto-title
    holds AT LEAST ``messages`` messages (the upsert is fire-and-forget from
    the UI's point of view, and the parked one lands when the hold ends).

    The poll pumps the browser (``page.wait_for_timeout``) rather than
    ``time.sleep``: the sync Playwright dispatcher only runs while the test
    thread is inside a Playwright call, so a sleeping poll starves a pending
    route interception — the re-run upsert would sit unanswered and the wait
    would time out on a browser that is actually fine."""
    deadline = time.monotonic() + timeout_s
    last: dict[str, Any] | None = None
    while time.monotonic() < deadline:
        last = _find_row(app_url, cookies, title)
        if last is not None and last["message_count"] >= messages:
            return last
        page.wait_for_timeout(200)
    raise AssertionError(f"no auto-saved row for {title!r} with {messages} msgs (last: {last!r})")


def _delete_chat(app_url: str, cookies: dict[str, str], chat_id: str) -> None:
    """Best-effort row cleanup (a 404 — already deleted — is fine)."""
    httpx.delete(f"{app_url}/api/chats/{chat_id}", timeout=10, cookies=cookies)


def _cleanup(
    app_url: str, cookies: dict[str, str], chat_id: str | None, title: str
) -> None:
    """Delete the row this test created — by id when the test got that far,
    by its unique auto-title otherwise (a FAILING test must not leak a row
    into the shared e2e database)."""
    if chat_id is None:
        leaked = _find_row(app_url, cookies, title)
        if leaked is None:
            return
        chat_id = str(leaked["id"])
    _delete_chat(app_url, cookies, chat_id)


def _ask(page: Page, question: str) -> float:
    """Send one turn and wait until the answer has fully landed (the
    ``done`` event restored the Send button). Returns the seconds the turn
    took — the calibration the ``HOLD_S`` comment cites."""
    started = time.monotonic()
    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")
    return time.monotonic() - started


def _watch_answer(page: Page) -> None:
    """Install the page-side clock (see ``_WATCH_ANSWER_JS``) BEFORE the send
    whose turn it stamps."""
    page.evaluate(_WATCH_ANSWER_JS, MOCK_ANSWER_MARKER)


def _answer_at(page: Page) -> float | None:
    """The wall-clock ms at which the answer text first reached the DOM
    (``None`` when the watch never fired)."""
    value = page.evaluate("() => window.__borAnswerAt")
    return None if value is None else float(value)


def _msg_count(post_data: str | None) -> int:
    """How many messages a save body carries — the request-level view of
    what the auto-save was trying to write."""
    try:
        messages = json.loads(post_data or "{}").get("messages")
    except json.JSONDecodeError:
        return -1
    return len(messages) if isinstance(messages, list) else -1


def _hold_first_chat_write(page: Page, hold_s: float = HOLD_S) -> list[dict[str, Any]]:
    """Hold the FIRST write to the saved-chat surface for ``hold_s`` seconds
    before letting it reach the server (the archive-upload suite's pattern),
    and log every write that passes through.

    Each entry is ``{kind, messages, started_ms, released_ms}`` with ``kind``
    in ``create``/``update``/``mint``; ``started_ms``/``released_ms`` are wall
    clock ms (comparable with the page's ``Date.now()``), which is what lets
    scenario 3 prove the done-time save point landed while the create was
    still parked."""
    log: list[dict[str, Any]] = []
    state = {"held": False, "seq": 0}

    def handle(route: Route) -> None:
        req = route.request
        if req.method not in ("POST", "PUT"):
            route.continue_()
            return
        started = time.time() * 1000
        entry = {
            "seq": state["seq"],
            "kind": "create" if req.method == "POST" else "update",
            "messages": _msg_count(req.post_data),
            "outcome": "held",
            "started_ms": started,
            "released_ms": started,
        }
        state["seq"] += 1
        if not state["held"]:
            state["held"] = True
            time.sleep(hold_s)  # the parked auto-save — nothing reaches the server yet
        entry["released_ms"] = time.time() * 1000
        log.append(entry)
        route.continue_()

    def handle_mint(route: Route) -> None:
        log.append(
            {
                "seq": state["seq"],
                "kind": "mint",
                "messages": -1,
                "outcome": "pass",
                "started_ms": time.time() * 1000,
                "released_ms": time.time() * 1000,
            }
        )
        state["seq"] += 1
        route.continue_()

    page.route(CHAT_WRITE_URL, handle)
    page.route(CHAT_SHARE_URL, handle_mint)
    return log


def _reject_chat_writes(
    page: Page,
) -> tuple[list[dict[str, Any]], Callable[[bool], None]]:
    """Answer every write to the saved-chat surface with an injected **422**
    (the incident's own rejection) and log every write that passes through.

    Returns the log and the arming switch: ``set_armed(False)`` lets the
    writes through (the Share's own upsert must reach the server) while the
    log keeps recording, so the ordering assertion covers the share too.

    Two deliberate choices:

    * a REJECTION, not a delay — a held request keeps a STALE body in flight
      that then races the share's own write for the row (see the module
      docstring), while a rejected save leaves the row exactly as the
      incident left it (behind the conversation) with no competing write;
    * a **422**, not a 5xx — phase 189 A4 retries a TRANSIENT failure only,
      so nothing re-saves behind the test's back and the ONLY thing that can
      bring the row up to date is the Share's own upsert (A2). The shared
      page is therefore a clean witness for A2, and the request log is the
      ordering proof."""
    log: list[dict[str, Any]] = []
    state = {"seq": 0, "armed": True}

    def handle(route: Route) -> None:
        req = route.request
        if req.method not in ("POST", "PUT"):
            route.continue_()
            return
        entry = {
            "seq": state["seq"],
            "kind": "create" if req.method == "POST" else "update",
            "messages": _msg_count(req.post_data),
            "outcome": "rejected" if state["armed"] else "pass",
        }
        state["seq"] += 1
        log.append(entry)
        if state["armed"]:
            route.fulfill(
                status=422,
                content_type="application/json",
                body='{"detail":[{"loc":["body"],"msg":"injected by phase 189 e2e"}]}',
            )
            return
        route.continue_()

    def handle_mint(route: Route) -> None:
        log.append(
            {
                "seq": state["seq"],
                "kind": "mint",
                "messages": -1,
                "outcome": "pass",
            }
        )
        state["seq"] += 1
        route.continue_()

    def set_armed(value: bool) -> None:
        state["armed"] = value

    page.route(CHAT_WRITE_URL, handle)
    page.route(CHAT_SHARE_URL, handle_mint)
    return log, set_armed


def _grant_clipboard(page: Page, app_url: str) -> None:
    """Grant the async-clipboard permissions (``http://127.0.0.1`` is a
    secure context, but headless Chromium still needs the grant before
    ``writeText`` resolves — otherwise the inline-link fallback fires)."""
    page.context.grant_permissions(
        ["clipboard-read", "clipboard-write"], origin=app_url
    )


def _click_share(page: Page) -> None:
    """Press the chat page's Share pill and wait for its outcome on the live
    region — the clipboard line or the inline-field fallback, whichever the
    context supports (the phase-51 contract)."""
    page.locator("#share-chat-btn").click()
    if page.evaluate("() => !!navigator.clipboard"):
        expect(page.locator("#send-status")).to_have_text(
            "Share link copied.", timeout=20_000
        )
    else:
        expect(page.locator("#send-status")).to_have_text(
            "Share link ready — copy it from the field.", timeout=20_000
        )


# ---------------------------------------------------------------------------
# The incident payload: the done frame of a 34-round / 33-tool turn
# ---------------------------------------------------------------------------


def _incident_source(i: int) -> dict[str, str]:
    """A real ``SourceRef`` at realistic column values (``source`` ≤ 120,
    ``path`` ≤ 1000, ``title`` ≤ 500 — the phase-83 mirrors of the
    ``documents`` columns)."""
    return {
        "source": "Homelab",
        "path": f"notes/deploy/kubernetes/component-{i:02d}.md",
        "title": f"Component {i:02d} — cluster setup, upgrades and rollback notes",
    }


def _incident_tool(i: int) -> dict[str, Any]:
    """A real ``ToolCall`` as the UI's ``toolAcc`` keeps it (the read the
    agent ran in round ``i``)."""
    return {
        "name": "read",
        "argument": f"Homelab/notes/deploy/kubernetes/component-{i:02d}.md",
        "truncated": False,
        "chars_shown": 4_200 + i,
        "chars_total": 4_200 + i,
        "offset": None,
        "limit": None,
    }


def _incident_brain() -> dict[str, Any]:
    """The incident turn's brain record: 22 ``sources`` (the READ docs —
    ``merge_cited_docs``), 33 ``tools``, 51 ``segments`` (the 33 tool entries
    + 18 thinking entries, one per contiguous thinking run of the 34-round
    ``xhigh`` turn)."""
    tools = [_incident_tool(i) for i in range(33)]
    segments: list[dict[str, Any]] = []
    for i, tool in enumerate(tools):
        if i < 18:  # a thinking run opens the first 18 rounds
            segments.append({"kind": "thinking", "text": f"round {i} scratchpad"})
        segments.append({"kind": "tool", **tool})
    return {
        "who": "brain",
        "text": "Everything I found, across all 22 documents I read.",
        "sources": [_incident_source(i) for i in range(22)],
        "related": [_incident_source(i) for i in range(2)],
        "deflected": False,
        "suggestions": [f"What about component {i}?" for i in range(5)],
        "thinking": " ".join(f"round {i}" for i in range(18)),
        "tools": tools,
        "segments": segments,
    }


def _assert_incident_lists(messages: list[dict[str, Any]]) -> None:
    """All three lists survived the round trip intact — nothing silently
    truncated on the way in, and the segment chronology is preserved."""
    brain = messages[1]
    assert len(brain["sources"]) == 22
    assert len(brain["related"]) == 2
    assert len(brain["tools"]) == 33
    assert len(brain["segments"]) == 51
    assert [s["path"] for s in brain["sources"]] == [
        _incident_source(i)["path"] for i in range(22)
    ]
    assert [t["argument"] for t in brain["tools"]] == [
        _incident_tool(i)["argument"] for i in range(33)
    ]
    kinds = [s["kind"] for s in brain["segments"]]
    assert kinds[:3] == ["thinking", "tool", "thinking"]
    assert kinds[-1] == "tool"
    assert kinds.count("thinking") == 18 and kinds.count("tool") == 33


# ---------------------------------------------------------------------------
# 1. A1 at the wire: the incident payload saves (201 / 200), anonymously
# ---------------------------------------------------------------------------


def test_incident_payload_shape_saves_at_the_wire(
    page: Page, app_url: str, db_ready: None
) -> None:
    """THE incident regression at the network layer: the done-time auto-save
    of a long turn — 22 cited READ docs, 33 tool calls, 51 timeline segments
    — is accepted anonymously (``POST`` **201**, the re-save ``PUT``
    **200**) and the row round-trips all three lists. Pre-phase-189 both
    calls were **422** at the round-cap-10 list caps, the quiet contract
    swallowed the failure and the row kept its pre-turn (question-only)
    state."""
    page.set_default_timeout(30_000)
    # Settled ANONYMOUS state — the write surface is public, and the
    # rejection happened on an anonymous browser save (the audit vector).
    page.goto(app_url + "/")
    expect(page.locator("#sign-in-link")).to_be_visible(timeout=15_000)

    question = "Audit my whole homelab and tell me what needs updating (phase 189 wire)"
    body = {
        "messages": [{"who": "user", "text": question}, _incident_brain()],
    }

    created = page.request.post(app_url + "/api/chats", data=body)
    assert created.status == 201, (
        f"the incident shape must save, not 422: {created.status} {created.text()}"
    )
    payload = created.json()
    chat_id = str(payload["id"])
    assert payload["message_count"] == 2
    _assert_incident_lists(payload["messages"])

    try:
        # The done-time auto-save's actual path: the re-save upsert of the
        # linked row (the incident's 422 rode exactly this request).
        updated = page.request.put(app_url + f"/api/chats/{chat_id}", data=body)
        assert updated.status == 200, (
            f"the incident shape must re-save, not 422: {updated.status} {updated.text()}"
        )
        _assert_incident_lists(updated.json()["messages"])

        # The row itself, through the admin read (the browser's own origin).
        login(page, app_url, next="/")
        cookies = _admin_cookies(page)
        got = _row(app_url, cookies, chat_id)
        assert got["message_count"] == 2
        _assert_incident_lists(got["messages"])
    finally:
        login(page, app_url, next="/")
        _delete_chat(app_url, _admin_cookies(page), chat_id)


# ---------------------------------------------------------------------------
# 2. A real mock turn's done-time auto-save lands (row: 2 messages)
# ---------------------------------------------------------------------------


def test_done_time_auto_save_lands_for_a_mock_turn(
    page: Page, app_url: str, kb: None, db_ready: None
) -> None:
    """The happy path through the real UI: one grounded mock turn, no
    interception at all. The conversation auto-saves at the save points and
    the row ends with BOTH messages, the brain record holding the answer the
    bubble showed (the baseline every other scenario here deviates from)."""
    page.set_default_timeout(30_000)
    login(page, app_url, next="/")
    expect(page).to_have_url(app_url + "/", timeout=30_000)

    q = "How is my Kubernetes cluster set up? (save-long-baseline)"
    turn_s = _ask(page, q)

    cookies = _admin_cookies(page)
    created: str | None = None
    try:
        row = _wait_row_messages(app_url, cookies, _auto_title(q), 2, page)
        created = str(row["id"])
        assert row["message_count"] == 2, "the send + done save points both landed"
        messages = _row(app_url, cookies, created)["messages"]
        assert [m["who"] for m in messages] == ["user", "brain"]
        assert messages[0]["text"] == q
        assert MOCK_ANSWER_MARKER in messages[1]["text"], (
            "the saved brain record holds the answer the bubble showed"
        )
        assert q[:40] in messages[1]["text"], (
            "the mock answer echoes the question — the saved text is THAT turn"
        )
        # The calibration HOLD_S cites: the mock turn settles long before the
        # hold ends, which is what makes scenario 3's overlap deterministic.
        assert turn_s < HOLD_S - 2.0, (
            f"the mock turn took {turn_s:.2f}s — HOLD_S ({HOLD_S}s) no longer "
            "clears it, so scenario 3's overlap would not be deterministic"
        )
    finally:
        _cleanup(app_url, cookies, created, _auto_title(q))


# ---------------------------------------------------------------------------
# 3. A3: the done-time save dropped by the in-flight guard RE-RUNS
# ---------------------------------------------------------------------------


def test_dropped_done_time_save_reruns_after_the_inflight_upsert_settles(
    page: Page, app_url: str, kb: None, db_ready: None
) -> None:
    """The owner-visible half of evidence 6: the send-time auto-save is
    parked at the server (held in the route), the mock turn settles inside
    that window, and the done-time save point it produces is refused by the
    one-upsert-at-a-time guard. Pre-phase-189 that save point VANISHED
    ("the next save point retries" — which never came) and the row stayed
    question-only; phase 189 A3 marks it PENDING and the parked upsert's
    ``finally`` re-runs it once it settles, so the row ends with BOTH
    messages."""
    page.set_default_timeout(30_000)
    login(page, app_url, next="/")
    expect(page).to_have_url(app_url + "/", timeout=30_000)

    _watch_answer(page)
    log = _hold_first_chat_write(page)

    q = "How is my Kubernetes cluster set up? (save-long-overlap)"
    _ask(page, q)  # the answer lands while the create is still parked

    cookies = _admin_cookies(page)
    created: str | None = None
    try:
        row = _wait_row_messages(app_url, cookies, _auto_title(q), 2, page)
        created = str(row["id"])
        # The owner-visible outcome: the conversation saved COMPLETE.
        assert row["message_count"] == 2, (
            "the dropped done-time save must re-run once the parked upsert settles"
        )
        messages = _row(app_url, cookies, created)["messages"]
        assert [m["who"] for m in messages] == ["user", "brain"]
        assert MOCK_ANSWER_MARKER in messages[1]["text"]

        # The mechanism, from the request log: ONE create carrying only the
        # question (the send-time save point), then exactly ONE update
        # carrying both messages (the pending re-run). Pre-phase-189 the
        # update never happened at all.
        assert [e["kind"] for e in log] == ["create", "update"], (
            f"expected the parked create then the single re-run, got {log}"
        )
        assert log[0]["messages"] == 1, "the send-time save held only the question"
        assert log[1]["messages"] == 2, "the re-run carried the brain record"

        # And the overlap is REAL, not a turn that happened to be slow: the
        # answer reached the DOM while the create was still parked, so the
        # done-time save point ran into the in-flight guard (a page-side
        # clock — the test thread cannot observe anything during the hold).
        answer_at = _answer_at(page)
        parked = log[0]
        assert answer_at is not None, "the page-side answer watch never fired"
        assert parked["started_ms"] < answer_at < parked["released_ms"], (
            f"the answer reached the DOM at {answer_at}, outside the window the "
            f"send-time save was parked "
            f"[{parked['started_ms']}, {parked['released_ms']}] — the save points "
            "did not overlap, so this test would not pin the re-run"
        )
    finally:
        _cleanup(app_url, cookies, created, _auto_title(q))


# ---------------------------------------------------------------------------
# 4. A2 (+A4 visible): Share saves the CURRENT conversation before the token
# ---------------------------------------------------------------------------


def test_share_saves_first_publishes_the_brain_message(
    page: Page,
    browser: Browser,
    app_url: str,
    kb: None,
    db_ready: None,
) -> None:
    """The incident's second half, through the real UI: the auto-saves do not
    land (an injected 422 — EXACTLY the incident's rejection), so the row
    stays behind the conversation, and the owner clicks Share. Pre-phase-189
    the Share minted its token on whatever the row last held, so the shared
    page showed ONLY the question; phase 189 A2 upserts the current
    conversation first (an in-cap ``PUT`` in the request log immediately
    before the ``POST …/share``) and the shared page carries the brain
    answer. A4 rides along visibly: the failed auto-saves leave the one-line
    note on screen (still no error banner, the phase-55 A2 shape)."""
    page.set_default_timeout(30_000)
    login(page, app_url, next="/")
    expect(page).to_have_url(app_url + "/", timeout=30_000)

    # Turn 1 saves cleanly, so the conversation is LINKED to a row (the
    # branch the incident rode: a linked Share, token-only pre-phase).
    q1 = "How is my Kubernetes cluster set up? (save-long-share-1)"
    _ask(page, q1)
    cookies = _admin_cookies(page)
    row = _wait_row_messages(app_url, cookies, _auto_title(q1), 2, page)
    chat_id = row["id"]

    anon_ctx: BrowserContext | None = None
    try:
        # The two auto-save points of turn 2 (send, brain-done) are refused:
        # the row stays at turn 1's state — the incident's stale row.
        log, set_armed = _reject_chat_writes(page)
        q2 = "How is my Kubernetes cluster set up? (save-long-share-2)"
        _ask(page, q2)

        rejected = [e for e in log if e["outcome"] == "rejected"]
        assert len(rejected) == 2, f"both turn-2 save points must be refused: {log}"
        assert [e["messages"] for e in rejected] == [3, 4], (
            f"the send point held 3 messages, the done point all 4: {log}"
        )
        # The row is BEHIND the conversation — the incident's precondition.
        assert _row(app_url, cookies, chat_id)["message_count"] == 2

        # A4: the failure is NOT silent any more — the one-line note is on
        # screen (and still no error banner, the phase-55 A2 shape).
        expect(page.locator("#send-status")).to_contain_text(SAVE_NOTE)
        expect(page.locator("#kb-banner")).to_be_hidden()

        # A2: the Share click saves FIRST, then mints. (The injection comes
        # off for the share itself — a share whose upsert fails ABORTS, and
        # that branch is marker-pinned in tests/unit/test_save_chat_ui.py.)
        set_armed(False)
        _grant_clipboard(page, app_url)
        _click_share(page)

        upserts = [e for e in log if e["kind"] == "update" and e["outcome"] == "pass"]
        mints = [e for e in log if e["kind"] == "mint"]
        assert len(mints) == 1, f"exactly one token mint: {log}"
        assert upserts, (
            "the linked Share must upsert the current conversation before minting"
        )
        assert upserts[-1]["seq"] < mints[-1]["seq"], (
            f"the share's upsert must precede the mint: {log}"
        )
        assert upserts[-1]["messages"] == 4, (
            "the share upsert carried the WHOLE conversation, answer included"
        )

        # The row caught up (2 turns = 4 messages) and carries the token.
        row = _wait_row_messages(app_url, cookies, _auto_title(q1), 4, page)
        assert row.get("share_url"), "the mint produced the share link"
        share_url = str(row["share_url"])

        # The shared page: the FULL conversation — the brain answer of turn 2
        # included (pre-phase-189: one question and nothing else).
        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(_auto_title(q1))
        expect(anon.locator(".msg.user .bubble")).to_have_count(2)
        bubbles = anon.locator(".msg.brain .bubble")
        expect(bubbles).to_have_count(2)
        expect(bubbles.nth(1)).to_contain_text(MOCK_ANSWER_MARKER, timeout=20_000)
        # The shared answer is turn 2's — the incident lost exactly this.
        expect(bubbles.nth(1)).to_contain_text("(save-long-share-2)")
        # Zero chips restored for a turn that read nothing (the phase-119
        # shape the shared page keeps).
        expect(anon.locator("a.source-chip")).to_have_count(0)
    finally:
        if anon_ctx is not None:
            anon_ctx.close()
        _delete_chat(app_url, cookies, chat_id)