feat(chat): save and view chat history — admin-only saved_chats, History page, open-a-chat return

tests/integration/test_api.py

@@ -88,6 +88,7 @@ def test_suggestions_honors_bor_suggestions_env_override(monkeypatch) -> None:
         ("/login.html", "Sign in"),  # phase 16: admin sign-in page
         ("/tuning.html", "Global Tuning"),  # phase 27: global tuning page
         ("/git-sources.html", "Git sources"),  # phase 35: admin git sources page
+        ("/history.html", "Saved chats"),  # phase 50: admin saved-chats page
     ],
 )
 def test_html_pages_served_locally_no_cdn(client, path: str, marker: str) -> None:
@@ -124,7 +125,8 @@ def test_index_page_no_cache_with_versioned_asset_refs(client) -> None:
 
 @pytest.mark.parametrize(
     "path",
-    ["/sources.html", "/document.html", "/login.html", "/tuning.html", "/git-sources.html"],
+    ["/sources.html", "/document.html", "/login.html", "/tuning.html",
+     "/git-sources.html", "/history.html"],  # phase 50: + the History page
 )
 def test_html_pages_no_cache_with_versioned_refs(client, path: str) -> None:
     """Each of the other four pages revalidates and carries at least one

tests/integration/test_chats_api.py

@@ -0,0 +1,446 @@
+"""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), and the delete 404/204.
+
+Requires:  podman compose up -d db
+"""
+from __future__ import annotations
+
+import time
+import uuid
+from collections.abc import Iterator
+from datetime import UTC, datetime, timedelta
+from typing import Any
+
+import pytest
+from fastapi.testclient import TestClient
+from sqlalchemy import select, text
+from sqlalchemy.orm import Session
+
+from app.main import app as fastapi_app
+from app.models import SavedChat
+
+FIRST_QUESTION = "How did I install gitlab?"
+EXPLICIT_TITLE = "My backup notes"
+
+#: 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"}
+ROW_KEYS = {"id", "title", "updated_at", "message_count"}
+
+
+@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()
+
+
+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),
+    ]
+    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

tests/integration/test_migration_0008.py

@@ -0,0 +1,255 @@
+"""Integration: migration 0008 (saved_chats) schema contract.
+
+Drives the **real Alembic engine** against the live dev database
+(``podman compose up -d db``), mirroring the house pattern of
+``test_migration_0005.py`` / ``test_migration_0007.py``
+(information_schema assertions on the state the migration must leave).
+The tests target revision ``0008`` explicitly so later migrations
+cannot break them:
+
+* upgrade 0007 → 0008 → a ``saved_chats`` table exists with
+  ``id UUID`` PK, ``title VARCHAR(500) NOT NULL``,
+  ``messages JSONB NOT NULL`` (the ``bor.chat.v1`` record list), and
+  ``created_at`` / ``updated_at TIMESTAMPTZ NOT NULL`` — both stamped
+  server-side by ``now()`` (an insert that omits them still lands
+  with both set);
+* the ``updated_at`` ORM ``onupdate`` bumps the timestamp on a row
+  update while ``created_at`` stays put (the phase-50 History page
+  orders by it);
+* downgrade to 0007 → the table is gone;
+* upgrade back to 0008 → it is back (round-trip).
+
+The ``alembic`` fixture guarantees the DB ends at head even if a test
+fails or the process is interrupted.
+"""
+from __future__ import annotations
+
+import json
+import time
+import uuid
+from collections.abc import Iterator
+from datetime import datetime
+from typing import Any
+
+import pytest
+from alembic.config import Config
+from sqlalchemy import text
+from sqlalchemy.orm import Session
+
+from alembic import command
+from app.db import db_available
+from app.models import SavedChat
+
+TITLE_BASE = "Mig 0008"
+MESSAGE_SHAPE = [{"who": "user", "text": "How did I install gitlab?"}]
+
+
+@pytest.fixture()
+def alembic(db: Session) -> Iterator[Config]:
+    """Real Alembic config bound to the dev DB (URL from app settings).
+
+    Starts at head (repairs an interrupted earlier run); teardown upgrades
+    to head no matter what happened, so the dev DB is never left below
+    head.
+    """
+    if not db_available():
+        pytest.skip("Postgres not reachable — run `podman compose up -d db` first")
+    cfg = Config()  # no alembic.ini file — env.py gets the URL from app config
+    cfg.set_main_option("script_location", "alembic")
+    command.upgrade(cfg, "head")
+    try:
+        yield cfg
+    finally:
+        command.upgrade(cfg, "head")
+
+
+def _table_exists(db: Session) -> bool:
+    """1 iff ``saved_chats`` is a table in this database."""
+    count: Any = db.execute(
+        text(
+            "SELECT count(*) FROM information_schema.tables"
+            " WHERE table_name = 'saved_chats'"
+        )
+    ).scalar()
+    assert count is not None, "information_schema count must be an int"
+    return int(count) == 1
+
+
+def _column(db: Session, column: str) -> tuple[Any, ...] | None:
+    """(data_type, is_nullable, column_default) for one saved_chats column."""
+    row = db.execute(
+        text(
+            "SELECT data_type, is_nullable, column_default"
+            " FROM information_schema.columns"
+            " WHERE table_name = 'saved_chats' AND column_name = :c"
+        ),
+        {"c": column},
+    ).fetchone()
+    return tuple(row) if row is not None else None
+
+
+def _pk_columns(db: Session) -> set[str]:
+    """Primary-key columns of ``saved_chats`` (empty if it does not exist)."""
+    rows = db.execute(
+        text(
+            "SELECT kcu.column_name"
+            " FROM information_schema.table_constraints tc"
+            " JOIN information_schema.key_column_usage kcu"
+            "   ON tc.constraint_name = kcu.constraint_name"
+            "  AND tc.table_schema = kcu.table_schema"
+            " WHERE tc.table_name = 'saved_chats'"
+            " AND tc.constraint_type = 'PRIMARY KEY'"
+        )
+    ).fetchall()
+    return {r[0] for r in rows}
+
+
+def _version(db: Session) -> str | None:
+    return db.execute(text("SELECT version_num FROM alembic_version")).scalar()
+
+
+def _raw_insert(db: Session, title: str) -> uuid.UUID:
+    """Insert one saved_chats row omitting the timestamps (server-stamped)."""
+    id: uuid.UUID = db.execute(
+        text(
+            "INSERT INTO saved_chats (id, title, messages)"
+            " VALUES (gen_random_uuid(), :t, CAST(:m AS jsonb))"
+            " RETURNING id"
+        ),
+        {"t": title, "m": json.dumps(MESSAGE_SHAPE)},
+    ).scalar_one()
+    db.commit()
+    return id
+
+
+def _delete(db: Session, id: uuid.UUID) -> None:
+    db.execute(text("DELETE FROM saved_chats WHERE id = :i"), {"i": id})
+    db.commit()
+
+
+def test_upgrade_to_0008_creates_saved_chats(db: Session, alembic: Config) -> None:
+    """Upgrade 0007 → 0008: ``saved_chats`` exists with the locked
+    columns, types, nullability, PK, and server ``now()`` defaults."""
+    command.downgrade(alembic, "0007")  # start from the pre-0008 state
+    assert _version(db) == "0007"
+    assert not _table_exists(db), "saved_chats must not exist before 0008"
+
+    command.upgrade(alembic, "0008")
+    assert _version(db) == "0008", "alembic_version must be at 0008"
+    assert _table_exists(db), "saved_chats is missing after 0008"
+
+    assert _pk_columns(db) == {"id"}, "saved_chats must have a single id PK"
+
+    id_col = _column(db, "id")
+    assert id_col is not None, "saved_chats.id is missing"
+    assert id_col[0] == "uuid", "saved_chats.id must be UUID"
+    assert id_col[1] == "NO", "saved_chats.id must be NOT NULL"
+
+    title = _column(db, "title")
+    assert title is not None, "saved_chats.title is missing"
+    assert title[0] == "character varying", "saved_chats.title must be VARCHAR"
+    assert title[1] == "NO", "saved_chats.title must be NOT NULL"
+
+    messages = _column(db, "messages")
+    assert messages is not None, "saved_chats.messages is missing"
+    assert messages[0] == "jsonb", "saved_chats.messages must be JSONB"
+    assert messages[1] == "NO", "saved_chats.messages must be NOT NULL"
+
+    for column in ("created_at", "updated_at"):
+        col = _column(db, column)
+        assert col is not None, f"saved_chats.{column} is missing"
+        assert col[0] == "timestamp with time zone", (
+            f"saved_chats.{column} must be TIMESTAMPTZ"
+        )
+        assert col[1] == "NO", f"saved_chats.{column} must be NOT NULL"
+        assert col[2] is not None and "now()" in col[2], (
+            f"saved_chats.{column} must default to now()"
+        )
+
+    # Phase 51 (share_token) must not leak into this minimal migration.
+    assert _column(db, "share_token") is None, (
+        "0008 stays minimal — share_token lands in 0009 (phase 51)"
+    )
+
+
+def test_server_timestamps_stamped_on_insert(db: Session, alembic: Config) -> None:
+    """An insert that omits created_at/updated_at (the API's shape) still
+    lands with both stamped by the server defaults."""
+    command.upgrade(alembic, "head")
+    chat_id = _raw_insert(db, f"{TITLE_BASE}: server stamps")
+    try:
+        created_at, updated_at = db.execute(
+            text("SELECT created_at, updated_at FROM saved_chats WHERE id = :i"),
+            {"i": chat_id},
+        ).one()
+        assert isinstance(created_at, datetime), "created_at must be server-stamped"
+        assert isinstance(updated_at, datetime), "updated_at must be server-stamped"
+        assert created_at.tzinfo is not None, "created_at must be timezone-aware"
+        # Fresh row: nothing has updated it, so both stamps agree (now).
+        assert (created_at - updated_at).total_seconds() < 5, (
+            "a fresh row must have created_at ≈ updated_at"
+        )
+    finally:
+        _delete(db, chat_id)
+
+
+def test_updated_at_bumps_on_row_update(db: Session, alembic: Config) -> None:
+    """The ORM ``onupdate=func.now()`` (the History page's Updated column)
+    bumps ``updated_at`` on a row update while ``created_at`` stays put."""
+    command.upgrade(alembic, "head")
+    chat = SavedChat(title=f"{TITLE_BASE}: before update", messages=MESSAGE_SHAPE)
+    db.add(chat)
+    db.commit()
+    try:
+        created_before: datetime = chat.created_at
+        updated_before: datetime = chat.updated_at
+        assert created_before is not None and updated_before is not None
+
+        time.sleep(0.1)  # now() has µs resolution — make the bump observable
+        chat.title = f"{TITLE_BASE}: after update"
+        chat.messages = [
+            {"who": "user", "text": "How did I install gitlab?"},
+            {"who": "brain", "text": "You've got this!", "sources": []},
+        ]
+        db.commit()
+        db.expire(chat)
+
+        created_after: datetime = chat.created_at
+        updated_after: datetime = chat.updated_at
+        assert created_after == created_before, "created_at must not move on update"
+        assert updated_after > updated_before, (
+            "updated_at must bump on a row update (onupdate=func.now())"
+        )
+    finally:
+        db.expire_all()
+        db.execute(text("DELETE FROM saved_chats WHERE id = :i"), {"i": chat.id})
+        db.commit()
+
+
+def test_downgrade_to_0007_drops_table(db: Session, alembic: Config) -> None:
+    """Downgrade to 0007: ``saved_chats`` is dropped (A13 — reversible)."""
+    command.downgrade(alembic, "0007")
+    assert _version(db) == "0007"
+    assert not _table_exists(db), "saved_chats must be dropped by the downgrade"
+    assert _column(db, "id") is None, "saved_chats.id must be gone"
+
+
+def test_upgrade_round_trip_restores_table(db: Session, alembic: Config) -> None:
+    """Downgrade to 0007, then upgrade back to 0008: the table is back
+    with its locked columns and PK."""
+    command.downgrade(alembic, "0007")
+    command.upgrade(alembic, "0008")
+    assert _version(db) == "0008", "round-trip upgrade must land at 0008"
+
+    assert _table_exists(db), "saved_chats must be back after the round-trip"
+    assert _pk_columns(db) == {"id"}, "saved_chats.id PK must be back"
+
+    messages = _column(db, "messages")
+    assert messages is not None and messages[0] == "jsonb", (
+        "saved_chats.messages must be JSONB after the round-trip"
+    )
+
+    col = _column(db, "updated_at")
+    assert col is not None and col[2] is not None and "now()" in col[2], (
+        "saved_chats.updated_at must keep its now() default after the round-trip"
+    )