feat(docs): save chat answers as docs — edit screen, commit + push to the .env docs branch

tests/integration/test_api.py

@@ -18,13 +18,16 @@ def test_health_reports_ok(client) -> None:
 
 
 def test_config_returns_default_app_metadata(client) -> None:
-    """GET /api/config is public (anonymous) and returns exactly two keys."""
+    """GET /api/config is public (anonymous) and returns exactly three
+    keys — the phase-39 app metadata + the phase-59 docs flag (inert
+    false while BOR_DOCS_REPO is empty — the "Save as doc" gating)."""
     r = client.get("/api/config")
     assert r.status_code == 200
     body = r.json()
-    assert set(body) == {"app_name", "version"}
+    assert set(body) == {"app_name", "version", "docs_repo_configured"}
     assert body["app_name"] == "Brain of Reese"
     assert body["version"] == get_settings().app_version
+    assert body["docs_repo_configured"] is False
 
 
 def test_config_follows_overridden_app_name(client) -> None:
@@ -39,9 +42,32 @@ def test_config_follows_overridden_app_name(client) -> None:
         r = client.get("/api/config")
         assert r.status_code == 200
         body = r.json()
-        assert set(body) == {"app_name", "version"}
+        assert set(body) == {"app_name", "version", "docs_repo_configured"}
         assert body["app_name"] == "Brain of Testy"
         assert body["version"] == "0.1.0"
+        assert body["docs_repo_configured"] is False
+    finally:
+        fastapi_app.dependency_overrides.clear()
+
+
+def test_config_docs_flag_tracks_settings(client) -> None:
+    """Phase 59 (task 05): ``docs_repo_configured`` mirrors
+    ``settings.docs_configured`` — a real bool (never a truthy string)
+    that flips true the moment BOR_DOCS_REPO is non-empty: that flag is
+    the entire frontend gating of the "Save as doc" button."""
+    from app.config import Settings
+    from app.main import app as fastapi_app
+
+    fastapi_app.dependency_overrides[get_settings] = lambda: Settings(
+        app_name="Brain of Testy",
+        docs_repo="/srv/docs-repo",
+    )
+    try:
+        r = client.get("/api/config")
+        assert r.status_code == 200
+        body = r.json()
+        assert isinstance(body["docs_repo_configured"], bool)
+        assert body["docs_repo_configured"] is True
     finally:
         fastapi_app.dependency_overrides.clear()
 

tests/integration/test_doc_drafts_api.py

@@ -0,0 +1,536 @@
+"""Integration: doc-drafts API (phase 59, task 02) — create / get / update.
+
+The draft lifecycle the edit screen runs on: create (from a response),
+fetch by token, update (modify before push) — all admin-only
+(router-wide ``require_admin``), all path-guard-railed (no path that can
+escape the repo root).
+
+Real Postgres (``podman compose up -d db``); no LLM involved — drafts
+are plain rows, so the suite is deterministic without a fake.
+
+Requires:  podman compose up -d db
+"""
+from __future__ import annotations
+
+import subprocess
+import uuid
+from collections.abc import Iterator
+from datetime import datetime
+from pathlib import Path
+from typing import Any
+
+import pytest
+from fastapi.testclient import TestClient
+from sqlalchemy import select, text
+
+from app.config import Settings, get_settings
+from app.main import app as fastapi_app
+from app.models import DocDraft
+
+TITLE = "How do I deploy a new service?"
+PATH = "docs/note.md"
+BODY = "# Answer\n\nSome **markdown** body."
+BASE_BRANCH = "main"
+DOCS_BRANCH = "bor-docs"
+
+
+def _settings(**kwargs: Any) -> Settings:
+    """Build Settings without reading a .env file (deterministic tests)."""
+    kwargs.setdefault("_env_file", None)
+    return Settings(**kwargs)  # pyright: ignore[reportCallIssue]  (kwarg exists at runtime)
+
+
+def _git_available() -> bool:
+    try:
+        proc = subprocess.run(["git", "--version"], capture_output=True, check=False)
+        return proc.returncode == 0
+    except (FileNotFoundError, OSError):
+        return False
+
+
+#: The push tests drive a real local git repo — skipped (not failed) on a
+#: machine without the git CLI (the task-03 unit-suite guard).
+GIT = _git_available()
+
+
+def _git(cwd: Path, *argv: str) -> str:
+    """Run git for the tests themselves (fixture setup + assertions)."""
+    proc = subprocess.run(["git", *argv], cwd=cwd, capture_output=True, text=True, check=False)
+    assert proc.returncode == 0, f"git {' '.join(argv)} failed: {proc.stderr}"
+    return proc.stdout
+
+
+@pytest.fixture(autouse=True)
+def clean_drafts(db) -> Iterator[None]:
+    """doc_drafts is global state: reset around every test."""
+    db.execute(text("TRUNCATE doc_drafts"))
+    db.commit()
+    yield
+    db.execute(text("TRUNCATE doc_drafts"))
+    db.commit()
+
+
+def _create(admin_client: TestClient, **overrides) -> dict:
+    """POST a well-formed draft (201) and return the response body."""
+    payload = {"title": TITLE, "path": PATH, "body": BODY, **overrides}
+    r = admin_client.post("/api/doc-drafts", json=payload)
+    assert r.status_code == 201, r.text
+    return r.json()
+
+
+def _backdate_updated_at(db, token: uuid.UUID) -> None:
+    """Push the row's ``updated_at`` one hour back (raw SQL — a hand-
+    written UPDATE does not trigger the column's onupdate default), so
+    a subsequent API write's bump is observable deterministically."""
+    db.execute(
+        text("UPDATE doc_drafts SET updated_at = now() - interval '1 hour' WHERE token = :t"),
+        {"t": token},
+    )
+    db.commit()
+
+
+# ---------- create (POST) ----------
+
+
+def test_create_returns_201_with_all_fields_and_draft_status(
+    admin_client: TestClient, db
+) -> None:
+    r = admin_client.post("/api/doc-drafts", json={"title": TITLE, "path": PATH, "body": BODY})
+
+    assert r.status_code == 201
+    body = r.json()
+    assert body["title"] == TITLE
+    assert body["path"] == PATH
+    assert body["body"] == BODY
+    assert body["status"] == "draft"
+    # The push feedback columns are NULL while still a draft.
+    assert body["branch"] is None
+    assert body["commit_sha"] is None
+    # The token: present, non-NULL, a valid (unguessable) UUID.
+    assert body["token"]
+    tok = uuid.UUID(body["token"])
+    assert body["created_at"]
+    assert body["updated_at"]
+
+    # The row is in Postgres under the same token (the URL credential).
+    row = db.execute(select(DocDraft).where(DocDraft.token == tok)).scalars().one()
+    assert row.title == TITLE
+    assert row.path == PATH
+    assert row.body == BODY
+    assert row.status == "draft"
+
+
+def test_create_strips_title_body_and_path(admin_client: TestClient) -> None:
+    body = _create(
+        admin_client, title=f"  {TITLE}  ", path=f"  {PATH} ", body=f"\n{BODY}\n"
+    )
+    assert body["title"] == TITLE
+    assert body["path"] == PATH
+    assert body["body"] == BODY
+
+
+def test_create_rejects_blank_title_body_path(admin_client: TestClient) -> None:
+    # Whitespace-only values: past pydantic's min_length=1, caught by the
+    # API's non-empty-after-strip rule (422), nothing stored.
+    for overrides in ({"title": "   "}, {"body": " \t\n "}, {"path": "   "}):
+        payload = {"title": TITLE, "path": PATH, "body": BODY, **overrides}
+        assert admin_client.post("/api/doc-drafts", json=payload).status_code == 422
+    # Truly empty title/body: pydantic 422 (min_length=1).
+    empty_title = {"title": "", "path": PATH, "body": BODY}
+    assert admin_client.post("/api/doc-drafts", json=empty_title).status_code == 422
+    empty_body = {"title": TITLE, "path": PATH, "body": ""}
+    assert admin_client.post("/api/doc-drafts", json=empty_body).status_code == 422
+
+
+# ---------- path guard-rails (shared with the push endpoint) ----------
+
+
+@pytest.mark.parametrize(
+    ("bad_path", "rule_in_detail"),
+    [
+        ("/etc/passwd", "absolute"),
+        ("../x.md", "'..'"),
+        ("a/b/../c.md", "'..'"),
+        ("no-suffix", "suffix"),
+        ("   ", "empty"),
+    ],
+)
+def test_create_path_guard_rejects_each_rule_with_422(
+    admin_client: TestClient, bad_path: str, rule_in_detail: str
+) -> None:
+    r = admin_client.post(
+        "/api/doc-drafts", json={"title": TITLE, "path": bad_path, "body": BODY}
+    )
+    assert r.status_code == 422
+    assert rule_in_detail in r.json()["detail"]
+
+
+def test_create_accepts_repo_relative_path_with_suffix(admin_client: TestClient) -> None:
+    body = _create(admin_client, path="docs/note.md")
+    assert body["path"] == "docs/note.md"
+
+
+def test_put_path_guard_rejects_traversal(admin_client: TestClient) -> None:
+    created = _create(admin_client)
+    for bad in ("/etc/passwd", "../x.md", "a/b/../c.md", "no-suffix"):
+        r = admin_client.put(
+            f"/api/doc-drafts/{created['token']}", json={"path": bad}
+        )
+        assert r.status_code == 422, bad
+    # The row is untouched by the rejected updates.
+    body = admin_client.get(f"/api/doc-drafts/{created['token']}").json()
+    assert body["path"] == PATH
+
+
+# ---------- get (by token) ----------
+
+
+def test_get_round_trips_created_draft(admin_client: TestClient) -> None:
+    created = _create(admin_client)
+
+    r = admin_client.get(f"/api/doc-drafts/{created['token']}")
+
+    assert r.status_code == 200
+    assert r.json() == created
+
+
+def test_get_unknown_token_returns_404(admin_client: TestClient) -> None:
+    r = admin_client.get(f"/api/doc-drafts/{uuid.uuid4()}")
+    assert r.status_code == 404
+    assert r.json() == {"detail": "draft not found"}
+
+
+def test_get_malformed_token_returns_422(admin_client: TestClient) -> None:
+    assert admin_client.get("/api/doc-drafts/not-a-uuid").status_code == 422
+
+
+# ---------- update (PUT) ----------
+
+
+def test_put_partial_body_only_keeps_title_and_path(admin_client: TestClient, db) -> None:
+    created = _create(admin_client)
+    token = uuid.UUID(created["token"])
+    _backdate_updated_at(db, token)
+
+    r = admin_client.put(f"/api/doc-drafts/{token}", json={"body": "# v2\n\nEdited."})
+
+    assert r.status_code == 200
+    body = r.json()
+    assert body["title"] == TITLE  # absent → unchanged
+    assert body["path"] == PATH  # absent → unchanged
+    assert body["body"] == "# v2\n\nEdited."
+    assert body["status"] == "draft"
+    assert body["created_at"] == created["created_at"]  # editing does not redate creation
+    # updated_at was bumped past the backdated value.
+    assert datetime.fromisoformat(body["updated_at"]) > datetime.fromisoformat(
+        created["updated_at"]
+    )
+
+
+def test_put_replaces_all_fields_when_supplied(admin_client: TestClient) -> None:
+    created = _create(admin_client)
+
+    r = admin_client.put(
+        f"/api/doc-drafts/{created['token']}",
+        json={"title": "New title", "path": "docs/other.md", "body": "New body."},
+    )
+
+    assert r.status_code == 200
+    body = r.json()
+    assert body["title"] == "New title"
+    assert body["path"] == "docs/other.md"
+    assert body["body"] == "New body."
+    assert body["status"] == "draft"
+
+
+def test_put_noop_still_bumps_updated_at(admin_client: TestClient, db) -> None:
+    """A PUT whose supplied values are all identical (or empty body)
+    changes no stored value — the ORM flushes nothing — yet the contract
+    is that a PUT bumps ``updated_at`` (the raw-UPDATE fallback)."""
+    created = _create(admin_client)
+    token = uuid.UUID(created["token"])
+    _backdate_updated_at(db, token)
+
+    r = admin_client.put(f"/api/doc-drafts/{token}", json={"body": BODY})  # identical
+
+    assert r.status_code == 200
+    assert r.json()["body"] == BODY
+    assert datetime.fromisoformat(r.json()["updated_at"]) > datetime.fromisoformat(
+        created["updated_at"]
+    )
+
+    # And an empty partial body (no fields at all) does the same.
+    _backdate_updated_at(db, token)
+    r2 = admin_client.put(f"/api/doc-drafts/{token}", json={})
+    assert r2.status_code == 200
+    assert datetime.fromisoformat(r2.json()["updated_at"]) > datetime.fromisoformat(
+        created["updated_at"]
+    )
+
+
+def test_put_resets_pushed_draft_to_draft(admin_client: TestClient, db) -> None:
+    created = _create(admin_client)
+    token = uuid.UUID(created["token"])
+
+    # Mark the draft pushed directly in the DB (the push endpoint's job
+    # lands in task 04 — here we pin the edit-side consequence).
+    row = db.execute(select(DocDraft).where(DocDraft.token == token)).scalars().one()
+    row.status = "pushed"
+    row.branch = "bor-docs"
+    row.commit_sha = "a" * 40
+    db.commit()
+
+    r = admin_client.put(f"/api/doc-drafts/{token}", json={"body": "Edited after push."})
+
+    assert r.status_code == 200
+    body = r.json()
+    assert body["status"] == "draft"  # the stored sha no longer describes the body
+    assert body["body"] == "Edited after push."
+    assert body["title"] == TITLE  # absent → unchanged
+    assert body["path"] == PATH  # absent → unchanged
+    # The last push stays visible until the next push overwrites it.
+    assert body["branch"] == "bor-docs"
+    assert body["commit_sha"] == "a" * 40
+
+
+def test_put_unknown_token_returns_404(admin_client: TestClient) -> None:
+    r = admin_client.put(f"/api/doc-drafts/{uuid.uuid4()}", json={"body": "x"})
+    assert r.status_code == 404
+    assert r.json() == {"detail": "draft not found"}
+
+
+def test_put_malformed_token_returns_422(admin_client: TestClient) -> None:
+    assert admin_client.put("/api/doc-drafts/not-a-uuid", json={"body": "x"}).status_code == 422
+
+
+def test_put_rejects_blank_fields_and_leaves_row_unchanged(admin_client: TestClient) -> None:
+    created = _create(admin_client)
+    for bad in ({"title": "   "}, {"body": " \t "}, {"path": "   "}):
+        assert admin_client.put(f"/api/doc-drafts/{created['token']}", json=bad).status_code == 422
+    assert admin_client.get(f"/api/doc-drafts/{created['token']}").json() == created
+
+
+# ---------- auth: anonymous gets 403 on every route ----------
+
+
+def test_anonymous_gets_403_on_all_routes(admin_client: TestClient, db) -> None:
+    created = _create(admin_client, body="admin-created")
+    anon = TestClient(fastapi_app)  # fresh jar: truly anonymous (no cookie)
+
+    r = anon.post("/api/doc-drafts", json={"title": "x", "path": "docs/x.md", "body": "y"})
+    assert r.status_code == 403
+    assert r.json() == {"detail": "admin only"}
+    assert anon.get(f"/api/doc-drafts/{created['token']}").status_code == 403
+    assert anon.put(f"/api/doc-drafts/{created['token']}", json={"body": "nope"}).status_code == 403
+
+    # The anonymous attempts changed nothing: exactly the admin's draft
+    # exists, untouched.
+    rows = db.execute(select(DocDraft)).scalars().all()
+    assert len(rows) == 1
+    assert rows[0].body == "admin-created"
+
+
+# ---------- push (POST /{token}/push — task 04) ----------
+#
+# The push tests run against a **real local bare repo** (the task-03
+# unit pattern) and inject the endpoint's settings via the app's
+# ``Depends(get_settings)`` override (the house pattern —
+# ``app/api/config.py`` takes ``settings: Settings = Depends(
+# get_settings)``). Result assertions read the bare repo's state
+# (``git show <branch>:<path>``, ``git rev-parse``), not the response
+# alone.
+
+
+@pytest.fixture()
+def bare_docs_repo(tmp_path: Path) -> Path:
+    """A bare origin seeded with one commit on ``main`` (``README.md``)."""
+    if not GIT:
+        pytest.skip("git is not available on this machine")
+    bare = tmp_path / "bare.git"
+    _git(tmp_path, "init", "--bare", str(bare))
+    seed = tmp_path / "seed"
+    _git(tmp_path, "clone", str(bare), str(seed))
+    (seed / "README.md").write_text("# docs\n", encoding="utf-8")
+    _git(seed, "checkout", "-B", BASE_BRANCH)
+    _git(
+        seed,
+        "-c", "commit.gpgsign=false",
+        "-c", "user.name=Test",
+        "-c", "user.email=t@example.com",
+        "add", "README.md",
+    )
+    _git(
+        seed,
+        "-c", "commit.gpgsign=false",
+        "-c", "user.name=Test",
+        "-c", "user.email=t@example.com",
+        "commit", "-m", "seed README",
+    )
+    _git(seed, "push", "origin", BASE_BRANCH)
+    return bare
+
+
+@pytest.fixture()
+def docs_push_settings(bare_docs_repo: Path, tmp_path: Path) -> Iterator[Settings]:
+    """Settings pointing at the fixture bare repo, injected into the
+    endpoint's settings dependency; the override is removed after the
+    test (no leak into other tests' settings)."""
+    settings = _settings(
+        docs_repo=str(bare_docs_repo),
+        docs_branch=DOCS_BRANCH,
+        docs_base_branch=BASE_BRANCH,
+        docs_work_dir=str(tmp_path / "docs-workdir"),
+    )
+    fastapi_app.dependency_overrides[get_settings] = lambda: settings
+    try:
+        yield settings
+    finally:
+        fastapi_app.dependency_overrides.pop(get_settings, None)
+
+
+def test_push_success_commits_and_records_branch_and_sha(
+    admin_client: TestClient, db, docs_push_settings: Settings, bare_docs_repo: Path
+) -> None:
+    created = _create(admin_client)
+    _backdate_updated_at(db, uuid.UUID(created["token"]))
+
+    r = admin_client.post(f"/api/doc-drafts/{created['token']}/push")
+
+    assert r.status_code == 200, r.text
+    body = r.json()
+    assert body["status"] == "pushed"
+    assert body["branch"] == DOCS_BRANCH
+    sha = body["commit_sha"]
+    assert len(sha) == 40
+    # The source of truth is the bare repo's state — not the response:
+    # the file landed on the branch at exactly the returned sha, with
+    # the draft's body, under the fixed per-invocation identity.
+    assert _git(bare_docs_repo, "rev-parse", DOCS_BRANCH).strip() == sha
+    assert _git(bare_docs_repo, "show", f"{DOCS_BRANCH}:{PATH}") == BODY
+    ident = _git(bare_docs_repo, "log", "-1", DOCS_BRANCH, "--format=%an <%ae>").strip()
+    assert ident == "Brain of Reese <bor@local>"
+    assert _git(bare_docs_repo, "log", "-1", DOCS_BRANCH, "--format=%s").strip() == (
+        f"docs: {TITLE}"
+    )
+
+    # The DB row records the outcome (status + branch + sha), and the
+    # GET endpoint reports it.
+    row = db.execute(
+        select(DocDraft).where(DocDraft.token == uuid.UUID(created["token"]))
+    ).scalars().one()
+    assert row.status == "pushed"
+    assert row.branch == DOCS_BRANCH
+    assert row.commit_sha == sha
+    got = admin_client.get(f"/api/doc-drafts/{created['token']}").json()
+    assert got["status"] == "pushed"
+    assert got["branch"] == DOCS_BRANCH
+    assert got["commit_sha"] == sha
+    # The success bumped updated_at (past the backdated value).
+    assert datetime.fromisoformat(got["updated_at"]) > datetime.fromisoformat(
+        created["updated_at"]
+    )
+
+
+def test_push_unconfigured_returns_409_naming_variable(admin_client: TestClient, db) -> None:
+    """Default settings (``docs_repo=""``) → 409 naming the variable;
+    the row stays a draft (D3: inert by default)."""
+    created = _create(admin_client)
+    fastapi_app.dependency_overrides[get_settings] = lambda: _settings()  # docs_repo=""
+    try:
+        r = admin_client.post(f"/api/doc-drafts/{created['token']}/push")
+    finally:
+        fastapi_app.dependency_overrides.pop(get_settings, None)
+
+    assert r.status_code == 409
+    assert r.json() == {"detail": "docs repo not configured (BOR_DOCS_REPO)"}
+    body = admin_client.get(f"/api/doc-drafts/{created['token']}").json()
+    assert body["status"] == "draft"
+    assert body["branch"] is None
+    assert body["commit_sha"] is None
+
+
+def test_push_non_repo_dir_returns_502_with_git_stderr(
+    admin_client: TestClient, db, tmp_path: Path
+) -> None:
+    """A configured repo that is not a git repo → 502 with git's stderr
+    in the detail (the ``GitSyncError`` → ``detail`` mapping);
+    the row stays a draft (only a success mutates)."""
+    plain = tmp_path / "not-a-repo"
+    plain.mkdir()
+    (plain / "file.txt").write_text("not a repo\n", encoding="utf-8")
+    fastapi_app.dependency_overrides[
+        get_settings
+    ] = lambda: _settings(
+        docs_repo=str(plain),
+        docs_branch=DOCS_BRANCH,
+        docs_base_branch=BASE_BRANCH,
+        docs_work_dir=str(tmp_path / "docs-workdir"),
+    )
+    created = _create(admin_client)
+    try:
+        r = admin_client.post(f"/api/doc-drafts/{created['token']}/push")
+    finally:
+        fastapi_app.dependency_overrides.pop(get_settings, None)
+
+    assert r.status_code == 502
+    detail = r.json()["detail"]
+    # git's stderr is surfaced (the clone refusal of a non-repo dir).
+    assert "failed" in detail
+    assert "fatal: repository" in detail
+    body = admin_client.get(f"/api/doc-drafts/{created['token']}").json()
+    assert body["status"] == "draft"
+    assert body["branch"] is None
+    assert body["commit_sha"] is None
+
+
+def test_push_unknown_token_returns_404(
+    admin_client: TestClient, docs_push_settings: Settings
+) -> None:
+    r = admin_client.post(f"/api/doc-drafts/{uuid.uuid4()}/push")
+    assert r.status_code == 404
+    assert r.json() == {"detail": "draft not found"}
+
+
+def test_push_rejects_bad_stored_path_with_422(
+    admin_client: TestClient, db, docs_push_settings: Settings
+) -> None:
+    """A row whose stored path no longer passes the guard-rails must not
+    be pushable (422 naming the rule — re-validated on push, task 02
+    helper); the row is untouched."""
+    bad = DocDraft(token=uuid.uuid4(), title=TITLE, path="../evil.md", body=BODY)
+    db.add(bad)
+    db.commit()
+    db.refresh(bad)
+
+    r = admin_client.post(f"/api/doc-drafts/{bad.token}/push")
+
+    assert r.status_code == 422
+    assert "'..'" in r.json()["detail"]
+    row = db.get(DocDraft, bad.id)
+    assert row is not None
+    assert row.status == "draft"
+    assert row.branch is None
+    assert row.commit_sha is None
+
+
+def test_push_anonymous_returns_403(
+    admin_client: TestClient,
+    db,
+    docs_push_settings: Settings,
+    bare_docs_repo: Path,
+) -> None:
+    created = _create(admin_client)
+    anon = TestClient(fastapi_app)  # fresh jar: truly anonymous (no cookie)
+
+    r = anon.post(f"/api/doc-drafts/{created['token']}/push")
+
+    assert r.status_code == 403
+    assert r.json() == {"detail": "admin only"}
+    # The anonymous push attempt changed nothing: no branch on the bare
+    # repo, the row is still a draft (guest reads 403 too).
+    assert _git(bare_docs_repo, "branch", "--list", DOCS_BRANCH).strip() == ""
+    assert anon.get(f"/api/doc-drafts/{created['token']}").status_code == 403
+    row = db.execute(
+        select(DocDraft).where(DocDraft.token == uuid.UUID(created["token"]))
+    ).scalars().one()
+    assert row.status == "draft"

tests/integration/test_migration_0011.py

@@ -0,0 +1,322 @@
+"""Integration: migration 0011 (doc_drafts) schema contract.
+
+Drives the **real Alembic engine** against the live dev database
+(``podman compose up -d db``), mirroring the house pattern of
+``test_migration_0010.py`` (information_schema / pg_indexes assertions
+on the state the migration must leave). The tests target revision
+``0011`` explicitly so later migrations cannot break them:
+
+* upgrade 0010 → 0011 → the ``doc_drafts`` table exists with the full
+  column contract (``id`` UUID PK; ``token`` UUID NOT NULL + the UNIQUE
+  index ``ix_doc_drafts_token`` — the URL credential; ``title`` /
+  ``path`` / ``body`` TEXT NOT NULL; ``status`` TEXT NOT NULL default
+  'draft'; ``branch`` / ``commit_sha`` TEXT NULL; ``created_at`` /
+  ``updated_at`` TIMESTAMPTZ NOT NULL default now());
+* inserted rows round-trip: an omitted ``status`` defaults to 'draft'
+  with NULL ``branch`` / ``commit_sha`` (the pre-push state) and both
+  timestamps are stamped server-side; explicit push-state values
+  round-trip verbatim;
+* two identical tokens are rejected by the unique index (the token is
+  a unique handle — the share-token precedent, phase 51);
+* downgrade to 0010 → the table and index are gone (A13 — reversible),
+  the rest of the schema (e.g. ``saved_chats.share_token``) survives;
+* upgrade back to 0011 → the table and the unique index are 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 uuid
+from collections.abc import Iterator
+from typing import Any
+
+import pytest
+from alembic.config import Config
+from sqlalchemy import text
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.orm import Session
+
+from alembic import command
+from app.db import db_available
+
+
+@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 _version(db: Session) -> str | None:
+    return db.execute(text("SELECT version_num FROM alembic_version")).scalar()
+
+
+def _table_exists(db: Session, table: str) -> bool:
+    count: Any = db.execute(
+        text(
+            "SELECT count(*) FROM information_schema.tables"
+            " WHERE table_schema = 'public' AND table_name = :t"
+        ),
+        {"t": table},
+    ).scalar()
+    assert count is not None, "information_schema count must be an int"
+    return int(count) == 1
+
+
+def _column(db: Session, table: str, column: str) -> tuple[Any, ...] | None:
+    """(data_type, is_nullable, column_default) for one table column."""
+    row = db.execute(
+        text(
+            "SELECT data_type, is_nullable, column_default"
+            " FROM information_schema.columns"
+            " WHERE table_name = :t AND column_name = :c"
+        ),
+        {"t": table, "c": column},
+    ).fetchone()
+    return tuple(row) if row is not None else None
+
+
+def _unique_token_index(db: Session) -> int:
+    """1 iff ``ix_doc_drafts_token`` exists as a UNIQUE index."""
+    count: Any = db.execute(
+        text(
+            "SELECT count(*) FROM pg_indexes"
+            " WHERE tablename = 'doc_drafts'"
+            " AND indexname = 'ix_doc_drafts_token'"
+        ),
+    ).scalar()
+    assert count is not None, "pg_indexes count must be an int"
+    is_unique: Any = db.execute(
+        text(
+            "SELECT indisunique FROM pg_index"
+            " WHERE indexrelid = (SELECT oid FROM pg_class WHERE relname = 'ix_doc_drafts_token')"
+        ),
+    ).scalar()
+    return int(count) if is_unique else 0
+
+
+def _insert(
+    db: Session,
+    *,
+    token: uuid.UUID | None = None,
+    status: str | None = None,
+    branch: str | None = None,
+    commit_sha: str | None = None,
+) -> uuid.UUID:
+    """Insert one doc_drafts row. ``status=None`` omits the column
+    (server-default path); a ``token`` is always supplied — the
+    migration carries no server default (the ORM/API supplies it)."""
+    cols = ["id", "token", "title", "path", "body"]
+    params: dict[str, Any] = {
+        "t": "Mig 0011",
+        "p": "docs/mig-0011.md",
+        "b": "# Phase 59 migration probe\n",
+    }
+    if token is not None:
+        params["tok"] = token
+    if status is not None:
+        cols.append("status")
+        params["s"] = status
+    if branch is not None:
+        cols.append("branch")
+        params["br"] = branch
+    if commit_sha is not None:
+        cols.append("commit_sha")
+        params["sha"] = commit_sha
+    sql = (
+        f"INSERT INTO doc_drafts ({', '.join(cols)}) VALUES ("
+        "gen_random_uuid(), :tok, :t, :p, :b"
+        + (", :s" if status is not None else "")
+        + (", :br" if branch is not None else "")
+        + (", :sha" if commit_sha is not None else "")
+        + ") RETURNING id"
+    )
+    draft_id: uuid.UUID = db.execute(text(sql), params).scalar_one()
+    db.commit()
+    return draft_id
+
+
+def _delete(db: Session, draft_id: uuid.UUID) -> None:
+    db.execute(text("DELETE FROM doc_drafts WHERE id = :i"), {"i": draft_id})
+    db.commit()
+
+
+def test_upgrade_to_0011_adds_doc_drafts(db: Session, alembic: Config) -> None:
+    """Upgrade 0010 → 0011: the table + the unique token index exist
+    with the full column contract; the table is absent at 0010."""
+    command.downgrade(alembic, "0010")  # start from the pre-0011 state
+    assert _version(db) == "0010"
+    assert not _table_exists(db, "doc_drafts"), "doc_drafts must be absent at 0010"
+    assert _unique_token_index(db) == 0, "the token index must be absent at 0010"
+
+    command.upgrade(alembic, "0011")
+    assert _version(db) == "0011", "alembic_version must be at 0011"
+    assert _table_exists(db, "doc_drafts"), "doc_drafts must exist at 0011"
+
+    id_col = _column(db, "doc_drafts", "id")
+    assert id_col is not None, "doc_drafts.id is missing"
+    assert id_col[0] == "uuid", "doc_drafts.id must be UUID"
+    assert id_col[1] == "NO", "doc_drafts.id must be NOT NULL (PK)"
+
+    token = _column(db, "doc_drafts", "token")
+    assert token is not None, "doc_drafts.token is missing"
+    assert token[0] == "uuid", "doc_drafts.token must be UUID"
+    assert token[1] == "NO", "doc_drafts.token must be NOT NULL (no un-drafted state)"
+    assert _unique_token_index(db) == 1, "the unique token index is missing"
+
+    for name in ("title", "path", "body"):
+        col = _column(db, "doc_drafts", name)
+        assert col is not None, f"doc_drafts.{name} is missing"
+        assert col[0] == "text", f"doc_drafts.{name} must be TEXT"
+        assert col[1] == "NO", f"doc_drafts.{name} must be NOT NULL"
+
+    status = _column(db, "doc_drafts", "status")
+    assert status is not None, "doc_drafts.status is missing"
+    assert status[0] == "text", "doc_drafts.status must be TEXT"
+    assert status[1] == "NO", "doc_drafts.status must be NOT NULL"
+    assert str(status[2]).startswith("'draft'"), (
+        "doc_drafts.status must have server default 'draft'"
+    )
+
+    for name in ("branch", "commit_sha"):
+        col = _column(db, "doc_drafts", name)
+        assert col is not None, f"doc_drafts.{name} is missing"
+        assert col[0] == "text", f"doc_drafts.{name} must be TEXT"
+        assert col[1] == "YES", f"doc_drafts.{name} must be NULL until pushed"
+
+    for name in ("created_at", "updated_at"):
+        col = _column(db, "doc_drafts", name)
+        assert col is not None, f"doc_drafts.{name} is missing"
+        assert col[0] == "timestamp with time zone", (
+            f"doc_drafts.{name} must be TIMESTAMPTZ"
+        )
+        assert col[1] == "NO", f"doc_drafts.{name} must be NOT NULL"
+        assert str(col[2]).startswith("now("), (
+            f"doc_drafts.{name} must have server default now()"
+        )
+
+
+def test_inserted_rows_round_trip_the_pre_push_and_pushed_states(
+    db: Session, alembic: Config
+) -> None:
+    """At 0011, an omitted status defaults to 'draft' with NULL
+    branch/commit_sha (the pre-push state) and both timestamps are
+    stamped server-side; explicit push-state values round-trip
+    verbatim."""
+    command.upgrade(alembic, "head")
+    draft_token = uuid.uuid4()
+    draft_id = _insert(db, token=draft_token)
+    pushed_token = uuid.uuid4()
+    pushed_id = _insert(
+        db,
+        token=pushed_token,
+        status="pushed",
+        branch="bor-docs",
+        commit_sha="a" * 40,
+    )
+    try:
+        row = db.execute(
+            text(
+                "SELECT token, status, branch, commit_sha, created_at, updated_at"
+                " FROM doc_drafts WHERE id = :i"
+            ),
+            {"i": draft_id},
+        ).fetchone()
+        assert row is not None, "the draft row must exist"
+        assert row[0] == draft_token, "the token must round-trip verbatim"
+        assert row[1] == "draft", "an omitted status must default to 'draft'"
+        assert row[2] is None and row[3] is None, (
+            "branch/commit_sha must be NULL before the push endpoint runs"
+        )
+        assert row[4] is not None and row[5] is not None, (
+            "created_at/updated_at must be stamped server-side"
+        )
+
+        pushed = db.execute(
+            text(
+                "SELECT status, branch, commit_sha FROM doc_drafts WHERE id = :i"
+            ),
+            {"i": pushed_id},
+        ).fetchone()
+        assert pushed is not None, "the pushed row must exist"
+        assert tuple(pushed) == ("pushed", "bor-docs", "a" * 40), (
+            "explicit push-state values must round-trip verbatim"
+        )
+    finally:
+        _delete(db, draft_id)
+        _delete(db, pushed_id)
+
+
+def test_unique_index_rejects_duplicate_tokens(db: Session, alembic: Config) -> None:
+    """Two identical tokens are rejected by the unique index — the
+    token is the unique URL credential (the share-token precedent,
+    phase 51); a distinct token still lands."""
+    command.upgrade(alembic, "head")
+    dup_token = uuid.uuid4()
+    first_id = _insert(db, token=dup_token)
+    other_id: uuid.UUID | None = None
+    try:
+        try:
+            _insert(db, token=dup_token)
+        except IntegrityError:
+            db.rollback()  # the aborted transaction must not leak
+        else:
+            pytest.fail("a duplicate doc_drafts.token must be rejected")
+
+        # A different token is fine — only the exact duplicate is unique.
+        other_id = _insert(db, token=uuid.uuid4())
+    finally:
+        _delete(db, first_id)
+        if other_id is not None:
+            _delete(db, other_id)
+
+
+def test_downgrade_to_0010_drops_the_table(db: Session, alembic: Config) -> None:
+    """Downgrade to 0010: the table and the unique index are gone
+    (A13 — reversible) while the rest of the schema survives."""
+    command.downgrade(alembic, "0010")
+    assert _version(db) == "0010"
+    assert not _table_exists(db, "doc_drafts"), "doc_drafts must be dropped"
+    assert _unique_token_index(db) == 0, "the token index must be dropped"
+
+    token_col = _column(db, "saved_chats", "share_token")
+    assert token_col is not None and token_col[0] == "uuid", (
+        "saved_chats.share_token must survive the downgrade"
+    )
+    meta = _column(db, "sources_meta", "version")
+    assert meta is not None and meta[0] == "integer", (
+        "sources_meta.version must survive the downgrade"
+    )
+
+
+def test_upgrade_round_trip_restores_the_table(db: Session, alembic: Config) -> None:
+    """Downgrade to 0010, then upgrade back to 0011: the table and the
+    unique index are back."""
+    command.downgrade(alembic, "0010")
+    command.upgrade(alembic, "0011")
+    assert _version(db) == "0011", "round-trip upgrade must land at 0011"
+
+    assert _table_exists(db, "doc_drafts"), "doc_drafts must be back"
+    assert _unique_token_index(db) == 1, "the unique token index must be back"
+
+    status = _column(db, "doc_drafts", "status")
+    assert status is not None and status[1] == "NO", (
+        "status must be TEXT NOT NULL after the round-trip"
+    )
+    assert str(status[2]).startswith("'draft'"), (
+        "status must default to 'draft' after the round-trip"
+    )