"""Integration: saved-chat CRUD (phase 50, task 02) — the ``/api/chats``
contract.

Real Postgres (``podman compose up -d db``). The router sits behind the
phase-16 ``require_admin`` gate exactly like ``/api/steering`` (the
house pattern of ``test_steering_api.py``): anonymous callers get 403 on
every route; the admin CRUD exercises the auto-title convention (first
user message, whitespace-collapsed, 120-char cap + the no-user-message
fallback), the list order (``updated_at desc, id desc``), the
full-payload round-trip (a ``bor.chat.v1``-shaped brain record carrying
``sources``/``thinking``/``tools``/``stopped`` survives losslessly),
the PUT upsert semantics (replacement + title-keep + title-set +
``updated_at`` bump), the delete 404/204, and (phase 53, task 03) the
sources-version stamp + ``stale`` flag: create and re-Save stamp the
row's ``sources_version``, list/detail expose ``stale`` (true iff the
stamp is behind the current generation — computed server-side), and
share/unshare stay version-immune.

Requires:  podman compose up -d db
"""
from __future__ import annotations

import re
import time
import uuid
from collections.abc import Iterator
from datetime import UTC, datetime, timedelta
from pathlib import Path
from typing import Any

import pytest
from fastapi.testclient import TestClient
from sqlalchemy import select, text
from sqlalchemy.orm import Session

from app.config import Settings
from app.main import app as fastapi_app
from app.models import SavedChat, SourcesMeta
from app.rag.sources_meta import bump_sources_version

FIRST_QUESTION = "How did I install gitlab?"
EXPLICIT_TITLE = "My backup notes"

#: The share link's shape: the page path + a canonical (lowercase) UUID.
SHARE_URL_RE = re.compile(
    r"^/shared/[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$"
)

#: The PUBLIC read's exact key set — no id, no timestamps, no token.
SHARED_OUT_KEYS = {"title", "messages"}

#: A full ``bor.chat.v1`` brain record (phase 14 shape) — every optional
#: key present; the round-trip test asserts it survives byte-identical.
FULL_BRAIN: dict[str, Any] = {
    "who": "brain",
    "text": "Your k3s cluster runs on three nodes — you've got this.",
    "sources": [
        {"source": "Homelab", "path": "kubernetes.md", "title": "Kubernetes Cluster"}
    ],
    "deflected": False,
    "suggestions": ["What ports does Traefik expose?"],
    "thinking": "The kubernetes doc covers the cluster layout…",
    "tools": [
        {"name": "read_document", "argument": "Homelab/kubernetes.md"},
        {"name": "list_documents", "argument": None},
    ],
    "stopped": False,
}

OUT_KEYS = {
    "id",
    "title",
    "created_at",
    "updated_at",
    "message_count",
    "messages",
    "stale",  # phase 53: server-computed staleness flag
}
ROW_KEYS = {"id", "title", "updated_at", "message_count", "stale"}


@pytest.fixture(autouse=True)
def clean_chats(db: Session) -> Iterator[None]:
    """``saved_chats`` is global state: reset around every test."""
    db.execute(text("TRUNCATE saved_chats"))
    db.commit()
    yield
    db.execute(text("TRUNCATE saved_chats"))
    db.commit()


@pytest.fixture(autouse=True)
def seeded_sources_meta(db: Session) -> Iterator[None]:
    """The ``sources_meta`` counter (phase 53) is global state: reset
    to the migration-0010 seed (id 1, version 0 — "the pre-counter
    KB") around every test, so each test starts from a known
    generation and the dev DB is left exactly as the migration left
    it."""
    db.execute(text("DELETE FROM sources_meta"))
    db.add(SourcesMeta(id=1, version=0))
    db.commit()
    yield
    db.execute(text("DELETE FROM sources_meta"))
    db.add(SourcesMeta(id=1, version=0))
    db.commit()


def _bump(db: Session) -> int:
    """One simulated KB-changing sync: a single committed bump
    (task 02's change-gate lives in the sync paths themselves; the API
    tests only need the counter's contract — caller commits).
    """
    version = bump_sources_version(db)
    db.commit()  # the helper only flushes
    return version


def _stored_sources_version(db: Session, chat_id: str) -> int:
    """The row's ``sources_version`` via raw SQL — deliberately
    bypassing the session's identity map, because the API's commits
    land in the app's own sessions (a cached ORM object could be
    stale)."""
    return int(
        db.execute(
            text("SELECT sources_version FROM saved_chats WHERE id = :id"),
            {"id": chat_id},
        ).scalar_one()
    )


def _user(text: str) -> dict[str, Any]:
    return {"who": "user", "text": text}


def _simple_conversation() -> list[dict[str, Any]]:
    return [_user(FIRST_QUESTION), {"who": "brain", "text": "You've got this!"}]


def _expect(records: list[dict[str, Any]]) -> list[dict[str, Any]]:
    """The stored shape of a record list (see app/api/chats.py):
    every record carries all ``bor.chat.v1`` keys, explicit nulls where
    an optional key does not apply (the restore path is null-safe).
    A record that already carries every key (``FULL_BRAIN``) is
    unchanged by this."""
    return [
        {
            "who": m["who"],
            "text": m["text"],
            "sources": m.get("sources"),
            "deflected": m.get("deflected"),
            "suggestions": m.get("suggestions"),
            "thinking": m.get("thinking"),
            "tools": m.get("tools"),
            "stopped": m.get("stopped"),
        }
        for m in records
    ]


def _assert_no_chats(admin_client: TestClient) -> None:
    assert admin_client.get("/api/chats").json() == {"chats": []}


# ---------- anonymous: 403 on every route (phase 16 gate) ----------


def test_anonymous_every_route_returns_403(client: TestClient) -> None:
    anon = TestClient(fastapi_app)  # fresh jar: truly anonymous
    unknown = uuid.uuid4()
    cases = [
        ("GET", "/api/chats", None),
        ("POST", "/api/chats", {"messages": _simple_conversation()}),
        ("GET", f"/api/chats/{unknown}", None),
        ("PUT", f"/api/chats/{unknown}", {"messages": _simple_conversation()}),
        ("DELETE", f"/api/chats/{unknown}", None),
        ("POST", f"/api/chats/{unknown}/share", None),
        ("POST", f"/api/chats/{unknown}/unshare", None),
    ]
    # The public read is NOT in this list — it is anonymous by design
    # (a wrong token 404s there, it never 403s).
    for method, path, body in cases:
        r = anon.request(method, path, json=body)
        assert r.status_code == 403, f"{method} {path} must be 403 for anonymous"
        assert r.json() == {"detail": "admin only"}


# ---------- create ----------


def test_create_returns_201_and_auto_titles(
    admin_client: TestClient, db: Session
) -> None:
    r = admin_client.post("/api/chats", json={"messages": _simple_conversation()})

    assert r.status_code == 201
    body = r.json()
    assert set(body) == OUT_KEYS
    assert body["title"] == FIRST_QUESTION  # auto-title = first user message
    assert body["message_count"] == 2
    assert body["messages"] == _expect(_simple_conversation())
    uuid.UUID(body["id"])  # valid UUID
    # Fresh row: nothing has updated it, so both stamps agree.
    assert body["created_at"] and body["updated_at"]
    assert abs(
        datetime.fromisoformat(body["created_at"])
        - datetime.fromisoformat(body["updated_at"])
    ).total_seconds() < 5
    rows = db.scalars(select(SavedChat)).all()
    assert [row.title for row in rows] == [FIRST_QUESTION]


def test_create_auto_title_collapses_whitespace_and_truncates_to_120(
    admin_client: TestClient,
) -> None:
    long_text = "How did I install " + "x" * 200
    r = admin_client.post(
        "/api/chats", json={"messages": [_user(long_text), {"who": "brain", "text": "ok"}]}
    )
    assert r.status_code == 201
    assert len(r.json()["title"]) == 120
    assert r.json()["title"] == long_text[:120]

    # Multi-space / tab / newline runs collapse to single spaces.
    r = admin_client.post(
        "/api/chats", json={"messages": [_user("What   is\nmy\tTraefik port?")]}
    )
    assert r.status_code == 201
    assert r.json()["title"] == "What is my Traefik port?"


def test_create_honors_explicit_title(admin_client: TestClient) -> None:
    r = admin_client.post(
        "/api/chats",
        json={"title": f"  {EXPLICIT_TITLE}  ", "messages": _simple_conversation()},
    )
    assert r.status_code == 201
    assert r.json()["title"] == EXPLICIT_TITLE  # trimmed, not auto-titled


def test_create_blank_title_falls_back_to_auto_title(admin_client: TestClient) -> None:
    r = admin_client.post(
        "/api/chats", json={"title": " \t\n ", "messages": _simple_conversation()}
    )
    assert r.status_code == 201
    assert r.json()["title"] == FIRST_QUESTION


def test_create_without_user_message_falls_back_to_chat_id(admin_client: TestClient) -> None:
    # Defensive — the UI cannot produce a conversation with no user
    # message; the auto-title then names the row after its own id.
    r = admin_client.post(
        "/api/chats", json={"messages": [{"who": "brain", "text": "hello"}]}
    )
    assert r.status_code == 201
    body = r.json()
    assert body["title"] == f"Chat {body['id'][:8]}"


def test_create_round_trips_full_brain_record(admin_client: TestClient) -> None:
    r = admin_client.post(
        "/api/chats", json={"messages": [_user(FIRST_QUESTION), FULL_BRAIN]}
    )
    assert r.status_code == 201
    # The bor.chat.v1-shaped payload round-trips losslessly: every
    # optional key (sources/deflected/suggestions/thinking/tools/
    # stopped) survives identical.
    assert r.json()["messages"][1] == FULL_BRAIN


def test_create_rejects_empty_messages(admin_client: TestClient) -> None:
    assert admin_client.post("/api/chats", json={"messages": []}).status_code == 422
    _assert_no_chats(admin_client)


def test_create_rejects_unknown_who(admin_client: TestClient) -> None:
    r = admin_client.post(
        "/api/chats", json={"messages": [{"who": "alien", "text": "hi"}]}
    )
    assert r.status_code == 422
    _assert_no_chats(admin_client)


def test_create_rejects_empty_text(admin_client: TestClient) -> None:
    assert (
        admin_client.post("/api/chats", json={"messages": [_user("")]})
    ).status_code == 422
    _assert_no_chats(admin_client)


def test_create_rejects_extra_message_keys(admin_client: TestClient) -> None:
    # A corrupted / HTML-shaped payload must not cross the boundary.
    message = _user(FIRST_QUESTION)
    message["html"] = "<b>not allowed</b>"
    assert admin_client.post("/api/chats", json={"messages": [message]}).status_code == 422
    _assert_no_chats(admin_client)


def test_create_rejects_title_over_500(admin_client: TestClient) -> None:
    assert (
        admin_client.post(
            "/api/chats", json={"title": "t" * 501, "messages": _simple_conversation()}
        )
    ).status_code == 422


# ---------- list ----------


def test_list_empty(admin_client: TestClient) -> None:
    r = admin_client.get("/api/chats")
    assert r.status_code == 200
    assert r.json() == {"chats": []}


def test_list_orders_by_updated_at_desc(admin_client: TestClient, db: Session) -> None:
    base = datetime.now(UTC)
    db.add_all(
        [
            SavedChat(
                title="oldest",
                messages=[_user("one")],
                updated_at=base,
            ),
            SavedChat(
                title="newest",
                messages=[_user("two"), _user("three")],
                updated_at=base + timedelta(hours=2),
            ),
            SavedChat(
                title="middle",
                messages=[_user("four")],
                updated_at=base + timedelta(hours=1),
            ),
        ]
    )
    db.commit()

    r = admin_client.get("/api/chats")
    assert r.status_code == 200
    body = r.json()
    assert [c["title"] for c in body["chats"]] == ["newest", "middle", "oldest"]
    for c in body["chats"]:
        assert set(c) == ROW_KEYS
        uuid.UUID(c["id"])
        assert "messages" not in c  # no payloads in the list


def test_list_reports_message_count(admin_client: TestClient) -> None:
    admin_client.post("/api/chats", json={"messages": _simple_conversation()})
    body = admin_client.get("/api/chats").json()
    assert [c["message_count"] for c in body["chats"]] == [2]


# ---------- get ----------


def test_get_returns_full_payload_round_trip(admin_client: TestClient) -> None:
    created = admin_client.post(
        "/api/chats",
        json={"title": EXPLICIT_TITLE, "messages": [_user(FIRST_QUESTION), FULL_BRAIN]},
    ).json()

    r = admin_client.get(f"/api/chats/{created['id']}")
    assert r.status_code == 200
    body = r.json()
    assert set(body) == OUT_KEYS
    assert body["id"] == created["id"]
    assert body["title"] == EXPLICIT_TITLE
    assert body["message_count"] == 2
    # Byte-identical payload: the brain record with sources/thinking/
    # tools/stopped (incl. the `argument: null` tool) survives the trip
    # to Postgres and back.
    assert body["messages"] == _expect([_user(FIRST_QUESTION), FULL_BRAIN])


def test_get_unknown_chat_returns_404(admin_client: TestClient) -> None:
    r = admin_client.get(f"/api/chats/{uuid.uuid4()}")
    assert r.status_code == 404
    assert r.json() == {"detail": "unknown chat"}


def test_get_invalid_id_returns_422(admin_client: TestClient) -> None:
    assert admin_client.get("/api/chats/not-a-uuid").status_code == 422


# ---------- update (PUT) — the re-Save upsert ----------


def test_put_replaces_messages_and_bumps_updated_at(admin_client: TestClient) -> None:
    created = admin_client.post(
        "/api/chats",
        json={"title": EXPLICIT_TITLE, "messages": _simple_conversation()},
    ).json()
    # A second, newer chat — it currently lists first.
    other = admin_client.post(
        "/api/chats", json={"messages": [_user("second question")], "title": "Other"}
    ).json()
    assert admin_client.get("/api/chats").json()["chats"][0]["id"] == other["id"]
    updated_before = created["updated_at"]

    time.sleep(0.1)  # now() has µs resolution — make the bump observable
    new_messages = [
        _user("How do I prune deleted docs?"),
        {"who": "brain", "text": "Use --prune."},
    ]
    r = admin_client.put(f"/api/chats/{created['id']}", json={"messages": new_messages})

    assert r.status_code == 200
    body = r.json()
    assert body["id"] == created["id"]
    assert body["title"] == EXPLICIT_TITLE  # absent title keeps the current one
    assert body["message_count"] == 2
    assert body["messages"] == _expect(new_messages)  # full replacement
    assert datetime.fromisoformat(body["created_at"]) == datetime.fromisoformat(
        created["created_at"]
    )  # editing does not redate creation
    assert datetime.fromisoformat(body["updated_at"]) > datetime.fromisoformat(
        updated_before
    ), "updated_at must bump on a re-Save (onupdate=func.now())"
    # The list order follows the bump: this row is first again.
    body_list = admin_client.get("/api/chats").json()["chats"]
    assert body_list[0]["id"] == created["id"]


def test_put_sets_title_when_supplied(admin_client: TestClient) -> None:
    created = admin_client.post(
        "/api/chats", json={"messages": _simple_conversation()}
    ).json()

    r = admin_client.put(
        f"/api/chats/{created['id']}",
        json={"title": "Renamed notes", "messages": _simple_conversation()},
    )

    assert r.status_code == 200
    assert r.json()["title"] == "Renamed notes"
    assert (
        admin_client.get(f"/api/chats/{created['id']}").json()["title"] == "Renamed notes"
    )


def test_put_blank_title_keeps_current(admin_client: TestClient) -> None:
    created = admin_client.post(
        "/api/chats",
        json={"title": EXPLICIT_TITLE, "messages": _simple_conversation()},
    ).json()

    r = admin_client.put(
        f"/api/chats/{created['id']}", json={"title": "  ", "messages": _simple_conversation()}
    )

    assert r.status_code == 200
    assert r.json()["title"] == EXPLICIT_TITLE


def test_put_unknown_chat_returns_404(admin_client: TestClient) -> None:
    r = admin_client.put(
        f"/api/chats/{uuid.uuid4()}", json={"messages": _simple_conversation()}
    )
    assert r.status_code == 404
    assert r.json() == {"detail": "unknown chat"}


def test_put_rejects_empty_messages(admin_client: TestClient) -> None:
    created = admin_client.post(
        "/api/chats", json={"messages": _simple_conversation()}
    ).json()
    assert (
        admin_client.put(f"/api/chats/{created['id']}", json={"messages": []})
    ).status_code == 422
    # The original payload is untouched.
    assert (
        admin_client.get(f"/api/chats/{created['id']}").json()["messages"]
        == _expect(_simple_conversation())
    )


def test_put_rejects_extra_message_keys(admin_client: TestClient) -> None:
    created = admin_client.post(
        "/api/chats", json={"messages": _simple_conversation()}
    ).json()
    bad = _user("hi")
    bad["innerHTML"] = "<script>alert(1)</script>"
    r = admin_client.put(
        f"/api/chats/{created['id']}", json={"messages": [bad, FULL_BRAIN]}
    )
    assert r.status_code == 422
    assert (
        admin_client.get(f"/api/chats/{created['id']}").json()["messages"]
        == _expect(_simple_conversation())
    )


# ---------- delete ----------


def test_delete_returns_204_and_removes(admin_client: TestClient, db: Session) -> None:
    created = admin_client.post("/api/chats", json={"messages": _simple_conversation()}).json()

    assert admin_client.delete(f"/api/chats/{created['id']}").status_code == 204
    assert admin_client.get(f"/api/chats/{created['id']}").status_code == 404
    assert admin_client.get("/api/chats").json() == {"chats": []}
    assert db.scalars(select(SavedChat)).all() == []


def test_delete_unknown_chat_returns_404(admin_client: TestClient) -> None:
    r = admin_client.delete(f"/api/chats/{uuid.uuid4()}")
    assert r.status_code == 404
    assert r.json() == {"detail": "unknown chat"}


def test_delete_invalid_id_returns_422(admin_client: TestClient) -> None:
    assert admin_client.delete("/api/chats/not-a-uuid").status_code == 422


# ---------- share / unshare / public read (phase 51, task 01) ----------


def _share(admin_client: TestClient, chat_id: str) -> dict[str, Any]:
    r = admin_client.post(f"/api/chats/{chat_id}/share")
    assert r.status_code == 200
    return r.json()


def _stored_token(db: Session, chat_id: str) -> uuid.UUID | None:
    """The row's ``share_token`` as seen by a fresh DB read."""
    row = db.get(SavedChat, uuid.UUID(chat_id))
    assert row is not None, "the chat row must exist"
    return row.share_token


def test_share_returns_200_with_share_url_and_is_idempotent(
    admin_client: TestClient, db: Session
) -> None:
    created = admin_client.post("/api/chats", json={"messages": _simple_conversation()}).json()

    body1 = _share(admin_client, created["id"])
    assert set(body1) == {"chat_id", "share_url"}
    assert body1["chat_id"] == created["id"]
    assert SHARE_URL_RE.fullmatch(body1["share_url"]), (
        f"share_url must be /shared/<lowercase uuid>: {body1['share_url']}"
    )
    token = uuid.UUID(body1["share_url"].removeprefix("/shared/"))
    # Persisted on the row (the A10 extension unchanged: same row, one
    # new column — no new table).
    assert _stored_token(db, created["id"]) == token

    # Idempotent: a re-share returns the SAME token, unchanged.
    body2 = _share(admin_client, created["id"])
    assert body2 == body1
    assert _stored_token(db, created["id"]) == token


def test_share_leaves_updated_at_unchanged(admin_client: TestClient) -> None:
    """Sharing is not a content edit — the token is written with a Core
    ``update()`` that skips the ORM ``onupdate``, so the History page's
    "latest activity first" order follows content edits only."""
    created = admin_client.post("/api/chats", json={"messages": _simple_conversation()}).json()
    updated_before = created["updated_at"]

    time.sleep(0.1)  # now() has µs resolution — make a bump observable
    _share(admin_client, created["id"])

    r = admin_client.get(f"/api/chats/{created['id']}")
    assert r.json()["updated_at"] == updated_before, (
        "share must not bump updated_at"
    )


def test_unshare_revokes_the_link_and_is_idempotent(
    admin_client: TestClient, db: Session
) -> None:
    created = admin_client.post("/api/chats", json={"messages": _simple_conversation()}).json()
    share_url = _share(admin_client, created["id"])["share_url"]
    anon = TestClient(fastapi_app)  # fresh jar: truly anonymous
    assert anon.get(f"/api{share_url}").status_code == 200  # live, pre-revoke

    r = admin_client.post(f"/api/chats/{created['id']}/unshare")
    assert r.status_code == 200
    assert r.json() == {"chat_id": created["id"], "shared": False}

    # The token is NULL in the DB and the public read now 404s.
    assert _stored_token(db, created["id"]) is None
    revoked = anon.get(f"/api{share_url}")
    assert revoked.status_code == 404
    assert revoked.json() == {"detail": "unknown or revoked share link"}

    # Idempotent: unsharing an unshared chat is a clean 200 (no write).
    r2 = admin_client.post(f"/api/chats/{created['id']}/unshare")
    assert r2.status_code == 200
    assert r2.json() == {"chat_id": created["id"], "shared": False}
    assert _stored_token(db, created["id"]) is None


def test_unshare_leaves_updated_at_unchanged(admin_client: TestClient) -> None:
    created = admin_client.post("/api/chats", json={"messages": _simple_conversation()}).json()
    updated_before = created["updated_at"]
    _share(admin_client, created["id"])

    time.sleep(0.1)
    admin_client.post(f"/api/chats/{created['id']}/unshare")

    r = admin_client.get(f"/api/chats/{created['id']}")
    assert r.json()["updated_at"] == updated_before, (
        "unshare must not bump updated_at"
    )


def test_public_read_returns_snapshot_without_private_keys(
    admin_client: TestClient,
) -> None:
    """A fresh anonymous client reads the shared chat: title + messages
    round-trip, and the body carries NONE of the admin-surface keys
    (no id, no timestamps, no token — a content snapshot, not a handle)."""
    created = admin_client.post(
        "/api/chats",
        json={
            "title": EXPLICIT_TITLE,
            "messages": [_user(FIRST_QUESTION), FULL_BRAIN],
        },
    ).json()
    share_url = _share(admin_client, created["id"])["share_url"]

    anon = TestClient(fastapi_app)  # fresh jar: truly anonymous
    r = anon.get(f"/api{share_url}")
    assert r.status_code == 200
    body = r.json()
    assert set(body) == SHARED_OUT_KEYS
    assert body["title"] == EXPLICIT_TITLE
    assert body["messages"] == _expect([_user(FIRST_QUESTION), FULL_BRAIN])


def test_public_read_wrong_and_revoked_tokens_404_with_one_detail(
    admin_client: TestClient,
) -> None:
    """Wrong (never issued) and revoked (unshared) tokens 404 with the
    SAME detail — no enumeration between the two cases."""
    anon = TestClient(fastapi_app)

    wrong = anon.get(f"/api/shared/{uuid.uuid4()}")
    assert wrong.status_code == 404
    assert wrong.json() == {"detail": "unknown or revoked share link"}

    created = admin_client.post("/api/chats", json={"messages": _simple_conversation()}).json()
    share_url = _share(admin_client, created["id"])["share_url"]
    assert admin_client.post(f"/api/chats/{created['id']}/unshare").status_code == 200
    revoked = anon.get(f"/api{share_url}")
    assert revoked.status_code == 404
    assert revoked.json() == wrong.json()  # one message, both cases


def test_public_read_malformed_token_returns_422() -> None:
    anon = TestClient(fastapi_app)
    assert anon.get("/api/shared/not-a-uuid").status_code == 422


def test_share_and_unshare_unknown_chat_return_404(admin_client: TestClient) -> None:
    unknown = uuid.uuid4()
    r = admin_client.post(f"/api/chats/{unknown}/share")
    assert r.status_code == 404
    assert r.json() == {"detail": "unknown chat"}
    r = admin_client.post(f"/api/chats/{unknown}/unshare")
    assert r.status_code == 404
    assert r.json() == {"detail": "unknown chat"}


# ---------- create-with-share (phase 51, task 02 — the save-then-share
# contract: one request saves AND shares; unshared shapes carry NO
# ``share_url`` key at all — absent, not null) ----------


def test_create_with_share_sets_token_in_the_same_commit(
    admin_client: TestClient, db: Session
) -> None:
    """``POST /api/chats`` with ``share: true``: the 201 body carries
    ``share_url`` (the ONLY extra key — the shape is OUT_KEYS +
    ``share_url``), matching the token shape, and the row's
    ``share_token`` is persisted in the SAME commit (one INSERT — no
    second request, no window where the row is saved but unshared)."""
    r = admin_client.post(
        "/api/chats", json={"messages": _simple_conversation(), "share": True}
    )
    assert r.status_code == 201
    body = r.json()
    assert set(body) == OUT_KEYS | {"share_url"}
    assert SHARE_URL_RE.fullmatch(body["share_url"]), (
        f"share_url must be /shared/<lowercase uuid>: {body['share_url']}"
    )
    token = uuid.UUID(body["share_url"].removeprefix("/shared/"))
    assert _stored_token(db, body["id"]) == token


def test_create_with_share_is_immediately_publicly_readable(
    admin_client: TestClient,
) -> None:
    """The save-then-share contract's payoff: the row is readable
    ANONYMOUSLY the moment the 201 lands (no second step)."""
    created = admin_client.post(
        "/api/chats", json={"messages": _simple_conversation(), "share": True}
    ).json()
    anon = TestClient(fastapi_app)  # fresh jar: truly anonymous
    r = anon.get(f"/api{created['share_url']}")
    assert r.status_code == 200
    body = r.json()
    assert set(body) == SHARED_OUT_KEYS
    assert body["messages"] == _expect(_simple_conversation())


def test_create_without_share_has_no_share_url(admin_client: TestClient) -> None:
    """The default (``share`` absent or false) is byte-for-byte the
    phase-50 shape: NO ``share_url`` key in the create body, the get
    body, or the list row — absent, not ``null``."""
    r = admin_client.post("/api/chats", json={"messages": _simple_conversation()})
    assert r.status_code == 201
    created = r.json()
    assert "share_url" not in created
    assert set(created) == OUT_KEYS
    got = admin_client.get(f"/api/chats/{created['id']}").json()
    assert "share_url" not in got and set(got) == OUT_KEYS
    row = admin_client.get("/api/chats").json()["chats"][0]
    assert "share_url" not in row and set(row) == ROW_KEYS


def test_create_share_false_is_explicitly_unshared(admin_client: TestClient) -> None:
    r = admin_client.post(
        "/api/chats", json={"messages": _simple_conversation(), "share": False}
    )
    assert r.status_code == 201
    assert "share_url" not in r.json(), "share: false is a plain Save (phase-50 shape)"


def test_list_rows_carry_share_url_only_when_shared(
    admin_client: TestClient,
) -> None:
    """The list endpoint populates ``share_url`` — so the History
    column renders straight from ``GET /api/chats`` (no second fetch
    per row): shared rows carry it (token shape), unshared rows omit it
    (the row shape is exactly ROW_KEYS)."""
    shared = admin_client.post(
        "/api/chats",
        json={"title": "Shared one", "messages": _simple_conversation(), "share": True},
    ).json()
    plain = admin_client.post(
        "/api/chats",
        json={"title": "Plain one", "messages": _simple_conversation()},
    ).json()
    rows = {c["id"]: c for c in admin_client.get("/api/chats").json()["chats"]}
    assert SHARE_URL_RE.fullmatch(rows[shared["id"]]["share_url"])
    assert set(rows[shared["id"]]) == ROW_KEYS | {"share_url"}
    assert "share_url" not in rows[plain["id"]]
    assert set(rows[plain["id"]]) == ROW_KEYS


def test_get_carry_share_url_and_unshare_drops_it(admin_client: TestClient) -> None:
    """``GET /{chat_id}`` carries ``share_url`` while shared (the same
    path as the create body) and drops the key after ``unshare`` — the
    full-payload shape returns to the phase-50 OUT_KEYS."""
    created = admin_client.post(
        "/api/chats", json={"messages": _simple_conversation(), "share": True}
    ).json()
    got = admin_client.get(f"/api/chats/{created['id']}").json()
    assert got["share_url"] == created["share_url"]
    assert set(got) == OUT_KEYS | {"share_url"}

    assert admin_client.post(f"/api/chats/{created['id']}/unshare").status_code == 200
    got2 = admin_client.get(f"/api/chats/{created['id']}").json()
    assert "share_url" not in got2
    assert set(got2) == OUT_KEYS


# ---------- sources-version stamp + stale flag (phase 53, task 03) ----------


def test_create_stamps_current_sources_version(
    admin_client: TestClient, db: Session
) -> None:
    """``POST`` stamps the PENDING row with the current generation —
    it ships in the same INSERT (the ``share_token`` precedent), and
    the 201 body reports ``stale: false`` (a fresh save is by
    definition current)."""
    created = admin_client.post(
        "/api/chats", json={"messages": _simple_conversation()}
    ).json()
    assert created["stale"] is False, "a fresh save is never stale"
    assert _stored_sources_version(db, created["id"]) == 0  # seed generation

    # After a KB-changing sync (one bump), the NEXT save stamps the
    # new generation and is still fresh.
    assert _bump(db) == 1
    created2 = admin_client.post(
        "/api/chats", json={"messages": [_user("another question")]}
    ).json()
    assert _stored_sources_version(db, created2["id"]) == 1
    assert created2["stale"] is False


def test_pre_counter_row_goes_stale_on_the_first_bump(
    admin_client: TestClient, db: Session
) -> None:
    """Recorded assumption: pre-existing rows (saved before the counter
    existed) stamp 0 — "the pre-counter KB" — and go stale on the
    first bump (0 < 1). A row inserted directly with the column's
    server default mirrors such a legacy row."""
    db.add(SavedChat(title="legacy", messages=[_user("old question")]))
    db.commit()  # no sources_version supplied → server default 0

    listing = admin_client.get("/api/chats").json()["chats"][0]
    assert listing["stale"] is False, "0 == 0: current at the pre-counter KB"

    assert _bump(db) == 1
    listing = admin_client.get("/api/chats").json()["chats"][0]
    assert listing["stale"] is True, "0 < 1: the first bump stale-s it"


def test_list_and_detail_report_stale_after_bump(admin_client: TestClient, db: Session) -> None:
    """The staleness flag is computed server-side in BOTH admin read
    shapes: after a bump, the list row and the detail payload of a
    row saved at the older generation report ``stale: true`` (and a
    second bump — still behind — stays stale)."""
    created = admin_client.post(
        "/api/chats", json={"messages": _simple_conversation()}
    ).json()
    assert created["stale"] is False

    assert _bump(db) == 1
    row = admin_client.get("/api/chats").json()["chats"][0]
    assert row["stale"] is True
    assert set(row) == ROW_KEYS  # stale sits in the standard row shape

    detail = admin_client.get(f"/api/chats/{created['id']}").json()
    assert detail["stale"] is True
    assert set(detail) == OUT_KEYS

    assert _bump(db) == 2  # still behind (stamp 0 < 2) → still stale
    assert admin_client.get(f"/api/chats/{created['id']}").json()["stale"] is True


def test_resave_restamps_and_clears_stale(admin_client: TestClient, db: Session) -> None:
    """A Re-Save (``PUT``) re-stamps the row to the CURRENT generation
    unconditionally — the owner is affirming this content against the
    current KB — so the 200 body reports ``stale: false`` again and
    the stored stamp advances (the manual escape hatch for a
    false-positive stale row)."""
    created = admin_client.post(
        "/api/chats", json={"messages": _simple_conversation()}
    ).json()
    assert _bump(db) == 1
    assert admin_client.get(f"/api/chats/{created['id']}").json()["stale"] is True

    r = admin_client.put(
        f"/api/chats/{created['id']}", json={"messages": _simple_conversation()}
    )
    assert r.status_code == 200
    assert r.json()["stale"] is False, "the re-Save affirms against generation 1"
    assert _stored_sources_version(db, created["id"]) == 1

    # A later bump stale-s it again — the stamp is a point in time, not
    # a sticky flag.
    assert _bump(db) == 2
    assert admin_client.get(f"/api/chats/{created['id']}").json()["stale"] is True


def test_share_and_unshare_leave_sources_version_untouched(
    admin_client: TestClient, db: Session
) -> None:
    """Share/unshare write ONLY ``share_token`` (raw SQL, the
    phase-51 contract) — the version stamp, like ``updated_at``, is
    immune: neither action can (un-)stale a chat, and staleness stays
    a pure function of the saved generation vs the current one."""
    assert _bump(db) == 1  # a non-zero stamp makes the assert observable
    created = admin_client.post(
        "/api/chats", json={"messages": _simple_conversation()}
    ).json()
    assert _stored_sources_version(db, created["id"]) == 1

    _share(admin_client, created["id"])
    assert _stored_sources_version(db, created["id"]) == 1
    assert admin_client.get(f"/api/chats/{created['id']}").json()["stale"] is False

    assert admin_client.post(f"/api/chats/{created['id']}/unshare").status_code == 200
    assert _stored_sources_version(db, created["id"]) == 1
    assert admin_client.get(f"/api/chats/{created['id']}").json()["stale"] is False


def test_public_read_snapshot_has_no_stale_surface(
    admin_client: TestClient, db: Session,
) -> None:
    """``/api/shared/<token>`` is a FROZEN snapshot by design (phase
    51): even after a bump, the anonymous body keeps exactly its
    title+messages key set — no ``stale`` flag, no staleness surface
    (an owner who regenerates can re-share afterwards)."""
    created = admin_client.post(
        "/api/chats", json={"messages": _simple_conversation(), "share": True}
    ).json()
    assert _bump(db) == 1  # the saved chat is now stale (admin surface)
    assert admin_client.get(f"/api/chats/{created['id']}").json()["stale"] is True

    anon = TestClient(fastapi_app)  # fresh jar: truly anonymous
    r = anon.get(f"/api{created['share_url']}")
    assert r.status_code == 200
    body = r.json()
    assert set(body) == SHARED_OUT_KEYS  # no stale key — frozen snapshot
    assert "stale" not in body


# ---------- the /shared/<token> page route (phase 51, task 01) ----------


def test_page_route_missing_shared_html_returns_same_404_json(
    client: TestClient, monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
    """Stale-deploy guard: a static dir WITHOUT ``shared.html`` (the
    page lands in task 03) 404s with the SAME JSON as the API — never a
    500, regardless of the token."""
    monkeypatch.setattr(
        "app.api.chats.get_settings", lambda: Settings(static_dir=str(tmp_path))
    )
    r = client.get(f"/shared/{uuid.uuid4()}")
    assert r.status_code == 404
    assert r.json() == {"detail": "unknown or revoked share link"}


def test_page_route_serves_shared_html_when_present(
    client: TestClient, monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
    """Once the file exists (task 03), the route serves it for any well-
    formed token — token validity is the page's own concern (it fetches
    the API and renders the "invalid or revoked" state itself)."""
    (tmp_path / "shared.html").write_text("<html>shared page</html>", encoding="utf-8")
    monkeypatch.setattr(
        "app.api.chats.get_settings", lambda: Settings(static_dir=str(tmp_path))
    )
    r = client.get(f"/shared/{uuid.uuid4()}")
    assert r.status_code == 200
    assert r.text == "<html>shared page</html>"
    assert "text/html" in r.headers["content-type"]