rfc-app/backend/app/api_branches.py

"""Slice 2 + Slice 4 API surface — the §8 active-RFC view and the §9.4
super-draft view share the same endpoint shape per §17's routing-collapse
rule. When `<slug>` resolves to an entry in state `super-draft`,
`<branch>` names a branch on the meta repo rather than on a per-RFC-repo
branch (§5 super-draft scoping note, §9.5). The dispatch happens here at
the API layer; the bot wrapper, the cache, and the chat layer all stay
state-agnostic — they take owner/repo/path arguments.

Owns every `branches/<branch>/...` and `threads/<thread_id>/...` route
from §17. Read paths fetch branch bodies live from Gitea (§4 #3 exempts
branch bodies from the cache); write paths funnel through `bot.py` so
the §1 chokepoint and the §6.5 trailer hold. The two §17 routes Slice 4
adds — `start-edit-branch` and `metadata` — live here too because they
are super-draft variants of the same machinery.

Visibility and contribute decisions are enforced inline here against the
§6 four-role model plus the §11 per-branch visibility/contribute rules;
the app's permission model is canonical, and Gitea sees only the bot.
"""
from __future__ import annotations

import asyncio
import json
import logging
import re
from typing import Any

from fastapi import APIRouter, HTTPException, Request
from fastapi.responses import StreamingResponse
from pydantic import BaseModel, Field

from . import auth, cache, chat as chat_layer, db, entry as entry_mod, funder, models_resolver, projects as projects_mod
from .bot import Bot
from .config import Config
from .gitea import Gitea, GiteaError
from .providers import BaseProvider

log = logging.getLogger(__name__)

RFC_FILE_PATH = "RFC.md"


# ---------------------------------------------------------------------------
# Request bodies
# ---------------------------------------------------------------------------


class PromoteToBranchBody(BaseModel):
    branch_name: str | None = Field(default=None, max_length=120)


class StartEditBranchBody(BaseModel):
    branch_name: str | None = Field(default=None, max_length=120)


class MetadataEditBody(BaseModel):
    title: str | None = Field(default=None, max_length=200)
    tags: list[str] | None = None
    pr_description: str | None = Field(default=None, max_length=4000)


class AcceptChangeBody(BaseModel):
    proposed: str = Field(min_length=0)
    was_edited_before_accept: bool = False
    force_apply_stale: bool = False


class ManualFlushBody(BaseModel):
    new_content: str
    paragraph_count: int = Field(ge=1)


class VisibilityBody(BaseModel):
    read_public: bool | None = None
    contribute_mode: str | None = Field(default=None, pattern="^(just-me|specific|any-contributor)$")


class GrantBody(BaseModel):
    grantee_gitea_login: str = Field(min_length=1, max_length=80)


class ThreadCreateBody(BaseModel):
    thread_kind: str = Field(pattern="^(chat|flag)$")
    anchor_kind: str = Field(pattern="^(whole-doc|range|paragraph)$")
    anchor_payload: dict | None = None
    label: str | None = Field(default=None, max_length=400)
    message: str | None = Field(default=None, max_length=20_000)


class ThreadMessageBody(BaseModel):
    text: str = Field(min_length=1, max_length=20_000)
    quote: str | None = Field(default=None, max_length=2000)


class ChatTurnBody(BaseModel):
    text: str = Field(min_length=1, max_length=20_000)
    quote: str | None = Field(default=None, max_length=2000)
    model: str | None = None


# ---------------------------------------------------------------------------
# Router
# ---------------------------------------------------------------------------


def make_router(
    config: Config,
    gitea: Gitea,
    bot: Bot,
    providers: dict[str, BaseProvider],
) -> APIRouter:
    router = APIRouter()

    # -------------------------------------------------------------------
    # §6.6 / §17: per-RFC model picker. The option list and default are
    # resolved per RFC by intersecting the meta-repo entry's optional
    # `models:` frontmatter with the operator's provisioned universe.
    # An empty resolved list means the RFC has opted out of AI per
    # §6.6 and the picker surfaces no options.
    # -------------------------------------------------------------------

    @router.get("/api/rfcs/{slug}/models")
    async def list_models_for_rfc(slug: str, request: Request) -> dict[str, Any]:
        # §22.5 visibility gate (subtractive): gated → 404 to non-members.
        _require_rfc(slug, auth.current_user(request))
        resolved = models_resolver.resolve_models_for_rfc(slug, providers)
        return {
            "models": [
                {"id": key, "name": providers[key].display_name}
                for key in resolved
            ],
            "default": resolved[0] if resolved else "",
        }

    # -------------------------------------------------------------------
    # §17: GET /api/rfcs/<slug>/main
    # For active RFCs: body, branches, open PRs on the per-RFC repo.
    # For super-drafts: canonical body (entry.body on meta-repo main),
    # open edit branches, open meta-repo body-edit and metadata PRs.
    # -------------------------------------------------------------------

    @router.get("/api/rfcs/{slug}/main")
    async def get_rfc_main(slug: str, request: Request) -> dict[str, Any]:
        viewer = auth.current_user(request)
        rfc = _require_rfc(slug, viewer)
        if rfc["state"] not in ("active", "super-draft"):
            raise HTTPException(409, f"RFC is {rfc['state']}")

        # Branches the viewer can read per §11.1. For active RFCs the
        # per-RFC repo's main is included so the §8.1 breadcrumb dropdown
        # can render it; for super-drafts the synthetic 'main' row that
        # `refresh_meta_branches` writes is internal scaffolding for the
        # §10.1 has-commits-ahead check — the §9.4 dropdown's first
        # position is rendered separately as 'canonical body'.
        if _is_meta_resident(rfc):
            branch_rows = db.conn().execute(
                """
                SELECT branch_name, head_sha, state, last_commit_at, pinned
                  FROM cached_branches
                 WHERE rfc_slug = ? AND state != 'deleted' AND branch_name != 'main'
                 ORDER BY last_commit_at DESC NULLS LAST
                """,
                (slug,),
            ).fetchall()
        else:
            branch_rows = db.conn().execute(
                """
                SELECT branch_name, head_sha, state, last_commit_at, pinned
                  FROM cached_branches
                 WHERE rfc_slug = ? AND state != 'deleted'
                 ORDER BY last_commit_at DESC NULLS LAST
                """,
                (slug,),
            ).fetchall()
        branches = [
            _branch_summary(slug, br, viewer)
            for br in branch_rows
            if _can_read_branch(slug, br["branch_name"], viewer)
        ]

        # Open PRs surfaced inline. For active: rfc_branch PRs on the
        # per-RFC repo. For super-draft: meta_body_edit and meta_metadata
        # PRs on the meta repo. Same shape either way — the §9.4 dropdown
        # treats both as "open work against this entry."
        pr_kinds = ("meta_body_edit", "meta_metadata") if _is_meta_resident(rfc) else ("rfc_branch",)
        placeholders = ",".join("?" * len(pr_kinds))
        pr_rows = db.conn().execute(
            f"""
            SELECT pr_number, title, state, head_branch, opened_by, opened_at, pr_kind
              FROM cached_prs
             WHERE rfc_slug = ? AND state = 'open' AND pr_kind IN ({placeholders})
             ORDER BY opened_at DESC
            """,
            (slug, *pr_kinds),
        ).fetchall()
        prs = [
            {
                "pr_number": r["pr_number"],
                "title": r["title"],
                "state": r["state"],
                "head_branch": r["head_branch"],
                "opened_by": r["opened_by"],
                "opened_at": r["opened_at"],
                "pr_kind": r["pr_kind"],
            }
            for r in pr_rows
        ]

        # For meta-resident entries the cached body is entry.body already
        # (see cache._upsert_cached_rfc), so no extraction is needed.
        # Pre-graduation history is a LEGACY-only affordance: under the
        # meta-only topology (§1, §13.4) graduation moves nothing, so an
        # active RFC's edit branches are its *current* branches and already
        # surface in `branches` above — there is no separate pre-graduation
        # set. The disclosure is therefore computed only for a legacy
        # per-RFC-repo active entry (`repo` set), where edit branches on the
        # meta repo genuinely predate the per-RFC repo and would otherwise
        # not appear. After the RFC-0001 fold-back (§13.6) nothing matches.
        pre_grad: list[dict[str, Any]] = []
        if rfc["state"] == "active" and rfc["repo"]:
            pre_grad_rows = db.conn().execute(
                """
                SELECT t.branch_name,
                       COUNT(DISTINCT t.id)        AS thread_count,
                       COUNT(DISTINCT m.id)        AS message_count,
                       MAX(m.created_at)           AS last_activity_at
                  FROM threads t
             LEFT JOIN thread_messages m ON m.thread_id = t.id
                 WHERE t.rfc_slug = ?
                   AND (
                        t.branch_name LIKE 'edit-' || ? || '-%'
                     OR t.branch_name LIKE 'edit/' || ? || '/%'
                   )
                 GROUP BY t.branch_name
                 ORDER BY MAX(m.created_at) DESC NULLS LAST, t.branch_name
                """,
                (slug, slug, slug),
            ).fetchall()
            for r in pre_grad_rows:
                change_count = db.conn().execute(
                    "SELECT COUNT(*) AS n FROM changes WHERE rfc_slug = ? AND branch_name = ?",
                    (slug, r["branch_name"]),
                ).fetchone()["n"]
                pre_grad.append({
                    "branch_name": r["branch_name"],
                    "thread_count": r["thread_count"],
                    "message_count": r["message_count"],
                    "change_count": change_count,
                    "last_activity_at": r["last_activity_at"],
                })

        return {
            "slug": slug,
            "title": rfc["title"],
            "state": rfc["state"],
            "id": rfc["rfc_id"],
            "repo": rfc["repo"],
            "tags": json.loads(rfc["tags_json"] or "[]"),
            "body": rfc["body"] or "",
            "body_sha": rfc["body_sha"],
            "branches": branches,
            "open_prs": prs,
            "pre_graduation_history": pre_grad,
        }

    # The bare `GET /api/rfcs/<slug>/branches/<branch>` is declared
    # at the *bottom* of this router so the more-specific deeper GET
    # routes — `branches/{branch:path}/threads` and
    # `branches/{branch:path}/threads/{thread_id}/messages` — match
    # before the bare GET swallows a sub-route path with `:path`'s
    # greedy match. Per the §19.2 "branch-name path routing" candidate
    # Slice 8 settles: ordering discipline against `{branch:path}` is
    # how the slashed-branch read works without collisions. See
    # `get_branch_view` below `stream_chat_turn`.

    # -------------------------------------------------------------------
    # §17: POST /api/rfcs/<slug>/branches/main/promote-to-branch
    # The §8.14 "Start Contributing on main" gesture for active RFCs.
    # Super-drafts use start-edit-branch below, per §9.5.
    # -------------------------------------------------------------------

    @router.post("/api/rfcs/{slug}/branches/main/promote-to-branch")
    async def promote_to_branch(slug: str, body: PromoteToBranchBody, request: Request) -> dict[str, Any]:
        viewer = auth.require_contributor(request)
        # v0.16.0 (item #12): cutting a contribute branch is the
        # PR-shaped write surface gate. A platform-granted user who is
        # not invited as a per-RFC contributor cannot start work that
        # only exists to land in a PR.
        if not auth.can_contribute_to_rfc(viewer, slug):
            raise HTTPException(
                403,
                "This RFC's owner has not invited you to contribute PRs",
            )
        rfc = _require_active_rfc(slug, viewer)
        owner, repo = _repo_for(rfc)
        new_branch = (body.branch_name or "").strip()
        if not new_branch:
            # Meta-only topology (§1): an active RFC's branches live on the
            # shared meta repo, so the auto name must embed the slug for the
            # cache to attribute it (`edit-<slug>-<hex>`, recovered by
            # `_slug_from_branch_name`). A legacy per-RFC-repo entry can use
            # the slug-free `<login>-draft-<hex>` since every branch there
            # belongs to the one RFC. The auto name is trusted (it carries a
            # reserved `edit-` prefix by design); only a user-supplied name
            # is validated, mirroring `start_edit_branch`.
            new_branch = (
                _auto_edit_branch_name(slug) if _is_meta_resident(rfc)
                else _auto_branch_name(viewer.gitea_login)
            )
        else:
            _validate_branch_name(new_branch)
        try:
            await bot.cut_branch_from_main(
                viewer.as_actor(),
                owner=owner,
                repo=repo,
                new_branch=new_branch,
                slug=slug,
            )
        except GiteaError as e:
            raise HTTPException(502, f"Gitea: {e.detail}")

        # Per §8.14, re-anchor any pending main-scoped changes by
        # mutating branch_name. They haven't been acted on yet, so
        # there is no audit trail to corrupt; `source_message_id`
        # continues to point at messages in main's chat — the schema
        # permits the cross-branch reference and the UI labels it as
        # "from a conversation on main."
        db.conn().execute(
            """
            UPDATE changes
               SET branch_name = ?
             WHERE rfc_slug = ? AND branch_name = 'main' AND state = 'pending'
            """,
            (new_branch, slug),
        )

        # Set the branch creator's default visibility (the spec
        # defaults already match, but we materialize the row so the
        # creator's identity travels with the branch).
        _ensure_branch_vis(slug, new_branch, creator_user_id=viewer.user_id)

        # Make the cache aware immediately so the breadcrumb reflects
        # the new branch without waiting for the webhook hop. Meta-resident
        # entries (§1) refresh meta branches; a legacy per-RFC repo refreshes
        # its own — `_refresh_cache_for` dispatches on residency.
        await _refresh_cache_for(rfc)

        return {"branch_name": new_branch, "slug": slug}

    # -------------------------------------------------------------------
    # §17 / §9.5: POST /api/rfcs/<slug>/start-edit-branch
    # The "Start Contributing" gesture on a super-draft — cuts a fresh
    # meta-repo branch the contributor will land body edits on.
    # -------------------------------------------------------------------

    @router.post("/api/rfcs/{slug}/start-edit-branch")
    async def start_edit_branch(slug: str, body: StartEditBranchBody, request: Request) -> dict[str, Any]:
        viewer = auth.require_contributor(request)
        # v0.16.0 (item #12): same per-RFC contribute gate as
        # promote-to-branch — kicking off a super-draft edit branch is
        # also PR-shaped work.
        if not auth.can_contribute_to_rfc(viewer, slug):
            raise HTTPException(
                403,
                "This RFC's owner has not invited you to contribute PRs",
            )
        rfc = _require_super_draft(slug, viewer)
        owner, repo = _repo_for(rfc)
        new_branch = (body.branch_name or "").strip()
        if not new_branch:
            new_branch = _auto_edit_branch_name(slug)
        else:
            _validate_branch_name(new_branch)
        try:
            await bot.cut_branch_from_main(
                viewer.as_actor(),
                owner=owner,
                repo=repo,
                new_branch=new_branch,
                slug=slug,
            )
        except GiteaError as e:
            raise HTTPException(502, f"Gitea: {e.detail}")

        # §9.6: re-anchor any pending main-scoped (super-draft canonical
        # body) changes onto the new edit branch, mirroring §8.14's
        # treatment for active RFCs.
        db.conn().execute(
            """
            UPDATE changes
               SET branch_name = ?
             WHERE rfc_slug = ? AND branch_name = 'main' AND state = 'pending'
            """,
            (new_branch, slug),
        )

        _ensure_branch_vis(slug, new_branch, creator_user_id=viewer.user_id)
        await cache.refresh_meta_branches(config, gitea)

        return {"branch_name": new_branch, "slug": slug}

    # -------------------------------------------------------------------
    # §17 / §9.5: POST /api/rfcs/<slug>/metadata
    # Title or tag edits on a super-draft — opens a tiny meta-repo PR
    # touching only the frontmatter of rfcs/<slug>.md. Slug renames are
    # not supported in v1 per §9.5 and the §19.2 candidate.
    # -------------------------------------------------------------------

    @router.post("/api/rfcs/{slug}/metadata")
    async def edit_metadata(slug: str, body: MetadataEditBody, request: Request) -> dict[str, Any]:
        viewer = auth.require_contributor(request)
        rfc = _require_super_draft(slug, viewer)
        # Permission: super-draft owners/arbiters per §6.3, plus app-wide
        # admins/owners per §6.1. Until claim, that collapses to admin/owner.
        if not _can_edit_metadata(rfc, viewer):
            raise HTTPException(403, "Only RFC owners/arbiters or app admins/owners may edit metadata")

        new_title = (body.title or "").strip() or None
        new_tags = body.tags

        if new_title is None and new_tags is None:
            raise HTTPException(422, "Provide title and/or tags")

        owner, repo = _repo_for(rfc)
        path = _file_path_for(rfc)
        fetched = await gitea.read_file(owner, repo, path, ref="main")
        if fetched is None:
            raise HTTPException(409, f"{path} not found on meta-main")
        prior_content, prior_sha = fetched
        try:
            entry = entry_mod.parse(prior_content)
        except Exception as e:
            raise HTTPException(500, f"meta-repo entry is malformed: {e}")

        changes_desc: list[str] = []
        if new_title is not None and new_title != entry.title:
            changes_desc.append(f"title: {entry.title!r} → {new_title!r}")
            entry.title = new_title
        if new_tags is not None and list(new_tags) != list(entry.tags):
            cleaned = [t.strip() for t in new_tags if t and t.strip()]
            changes_desc.append(f"tags: {entry.tags!r} → {cleaned!r}")
            entry.tags = cleaned

        if not changes_desc:
            return {"ok": True, "noop": True}

        new_content = entry_mod.serialize(entry)
        pr_title = f"Metadata: {entry.title}"
        pr_description = (
            body.pr_description
            or "Metadata edit on the super-draft entry:\n\n- " + "\n- ".join(changes_desc)
        )

        try:
            pr = await bot.open_metadata_pr(
                viewer.as_actor(),
                org=owner,
                meta_repo=repo,
                slug=slug,
                new_file_contents=new_content,
                prior_sha=prior_sha,
                pr_title=pr_title,
                pr_description=pr_description,
            )
        except GiteaError as e:
            raise HTTPException(502, f"Gitea: {e.detail}")

        await cache.refresh_meta_branches(config, gitea)
        await cache.refresh_meta_pulls(config, gitea)
        return {"pr_number": pr["number"], "slug": slug, "branch_name": pr["head"]["ref"]}

    # -------------------------------------------------------------------
    # §17 / §8.9: accept / decline / reask a change
    # -------------------------------------------------------------------

    @router.post("/api/rfcs/{slug}/branches/{branch:path}/changes/{change_id}/accept")
    async def accept_change(
        slug: str,
        branch: str,
        change_id: int,
        body: AcceptChangeBody,
        request: Request,
    ) -> dict[str, Any]:
        viewer = auth.require_contributor(request)
        rfc = _require_rfc_with_repo(slug, viewer)
        _require_can_contribute(slug, branch, viewer)
        row = _require_pending_change(slug, branch, change_id)
        if row["kind"] != "ai":
            raise HTTPException(409, "Manual changes are accepted via manual-flush")

        # Fetch current file and extract the editable body. For super-draft
        # the file is rfcs/<slug>.md with frontmatter; for active it's RFC.md.
        owner, repo = _repo_for(rfc, branch)
        path = _file_path_for(rfc, branch)
        fetched = await gitea.read_file(owner, repo, path, ref=branch)
        if fetched is None:
            raise HTTPException(409, f"Branch {path} not found")
        prior_content, prior_sha = fetched
        current_body = _extract_body(rfc, prior_content, branch)

        original = row["original"]
        occurrences = current_body.count(original)
        if occurrences == 0:
            if not body.force_apply_stale:
                # Per §8.11: mark stale and refuse.
                db.conn().execute(
                    "UPDATE changes SET stale_since = COALESCE(stale_since, datetime('now')) WHERE id = ?",
                    (change_id,),
                )
                raise HTTPException(409, "Change is stale — original text no longer in document")
            # force-apply path: append the proposed text at the end as a
            # coarse fallback. The contributor's explicit consent is the gate.
            new_body = current_body.rstrip() + "\n\n" + body.proposed.strip() + "\n"
        elif occurrences > 1:
            raise HTTPException(409, "Change cannot be auto-applied: original text appears multiple times")
        else:
            new_body = current_body.replace(original, body.proposed, 1)

        new_file_contents = _wrap_body(rfc, prior_content, new_body, branch)

        try:
            sha = await bot.commit_accepted_change(
                viewer.as_actor(),
                owner=owner,
                repo=repo,
                branch=branch,
                file_path=path,
                new_content=new_file_contents,
                prior_sha=prior_sha,
                change_id=change_id,
                original=original,
                proposed=body.proposed,
                ai_proposed=row["proposed"] if body.was_edited_before_accept else None,
                reason=row["reason"] or "",
                source_message_id=row["source_message_id"],
                slug=slug,
            )
        except GiteaError as e:
            raise HTTPException(502, f"Gitea: {e.detail}")

        db.conn().execute(
            """
            UPDATE changes
               SET state = 'accepted',
                   proposed = ?,
                   was_edited_before_accept = ?,
                   acted_by = ?, acted_at = datetime('now'),
                   commit_sha = ?,
                   stale_since = NULL
             WHERE id = ?
            """,
            (
                body.proposed,
                1 if body.was_edited_before_accept else 0,
                viewer.user_id,
                sha,
                change_id,
            ),
        )

        # Per §8.11: mark any pending AI proposals whose anchor no longer
        # locates as stale. The stale check operates against the editable
        # body, not the full file.
        chat_layer.mark_stale_overlapping(rfc_slug=slug, branch_name=branch, new_body=new_body)
        await _refresh_cache_for(rfc)

        return {"ok": True, "commit_sha": sha, "change_id": change_id}

    @router.post("/api/rfcs/{slug}/branches/{branch:path}/changes/{change_id}/decline")
    async def decline_change(slug: str, branch: str, change_id: int, request: Request) -> dict[str, Any]:
        viewer = auth.require_contributor(request)
        _require_rfc_with_repo(slug, viewer)
        _require_can_contribute(slug, branch, viewer)
        row = _require_pending_change(slug, branch, change_id)
        if row["kind"] != "ai":
            raise HTTPException(409, "Manual changes are declined via manual-flush revert")
        db.conn().execute(
            """
            UPDATE changes
               SET state = 'declined', acted_by = ?, acted_at = datetime('now')
             WHERE id = ?
            """,
            (viewer.user_id, change_id),
        )
        return {"ok": True, "change_id": change_id}

    @router.post("/api/rfcs/{slug}/branches/{branch:path}/changes/{change_id}/reask")
    async def reask_change(slug: str, branch: str, change_id: int, request: Request) -> dict[str, Any]:
        viewer = auth.require_contributor(request)
        rfc = _require_rfc_with_repo(slug, viewer)
        _require_can_contribute(slug, branch, viewer)
        row = _require_change(slug, branch, change_id)
        if row["kind"] != "ai":
            raise HTTPException(409, "Only AI changes can be reasked")
        thread_id = row["thread_id"]
        if thread_id is None:
            raise HTTPException(409, "Change has no originating thread")
        # §6.6: refuse cleanly if the RFC's resolved model list is empty —
        # either the operator has no providers, or the RFC opted out
        # (`models: []`), or the entry names only models the operator
        # no longer provisions. Same shape; same honest refusal.
        resolved = models_resolver.resolve_models_for_rfc(slug, providers)
        if not resolved:
            raise HTTPException(503, "No AI providers configured")
        reask_model = resolved[0]

        owner, repo = _repo_for(rfc, branch)
        path = _file_path_for(rfc, branch)
        fetched = await gitea.read_file(owner, repo, path, ref=branch)
        body_text = _extract_body(rfc, fetched[0], branch) if fetched else ""

        # §6.7: pick the funder-credentialed instance when a consenting
        # funder is in effect on this RFC; otherwise the operator's.
        provider = funder.provider_for_rfc(slug, reask_model, providers)
        if provider is None:
            raise HTTPException(503, "No AI providers configured")
        system = chat_layer.build_system_prompt(title=rfc["title"], body=body_text)
        history = chat_layer.build_history(thread_id)
        reask_prompt = (
            "The earlier proposal's `<original>` text no longer matches the document — "
            "the contributor has edited that passage. Please regenerate your proposal "
            "anchored to the current phrasing. Earlier <original>:\n\n"
            f"{row['original']}\n\nEarlier <proposed>:\n\n{row['proposed']}"
        )
        user_id = chat_layer.append_user_message(
            thread_id=thread_id, author_user_id=viewer.user_id, text=reask_prompt, quote=None
        )
        assistant_id = chat_layer.append_assistant_placeholder(
            thread_id=thread_id, model_id=reask_model
        )

        text = provider.send(system, history + [{"role": "user", "content": reask_prompt}])
        chat_layer.finalize_assistant_message(message_id=assistant_id, text=text)
        parsed = chat_layer.parse_changes(text)
        new_ids = chat_layer.materialize_changes(
            rfc_slug=slug,
            branch_name=branch,
            thread_id=thread_id,
            source_message_id=assistant_id,
            parsed=parsed,
        )
        return {
            "ok": True,
            "user_message_id": user_id,
            "assistant_message_id": assistant_id,
            "new_change_ids": new_ids,
            "assistant_text": text,
        }

    # -------------------------------------------------------------------
    # §17 / §8.11 / §10.6: manual-edit flush
    # -------------------------------------------------------------------

    @router.post("/api/rfcs/{slug}/branches/{branch:path}/manual-flush")
    async def manual_flush(slug: str, branch: str, body: ManualFlushBody, request: Request) -> dict[str, Any]:
        viewer = auth.require_contributor(request)
        rfc = _require_rfc_with_repo(slug, viewer)
        _require_can_contribute(slug, branch, viewer)
        owner, repo = _repo_for(rfc, branch)
        path = _file_path_for(rfc, branch)

        fetched = await gitea.read_file(owner, repo, path, ref=branch)
        if fetched is None:
            raise HTTPException(409, f"Branch {path} not found")
        prior_content, prior_sha = fetched
        prior_body = _extract_body(rfc, prior_content, branch)
        if prior_body == body.new_content:
            return {"ok": True, "noop": True}

        new_file_contents = _wrap_body(rfc, prior_content, body.new_content, branch)

        # Per §8.11: materialize the manual change as a `changes` row
        # first so the resolved card binds 1:1 to the commit.
        cur = db.conn().execute(
            """
            INSERT INTO changes
                (rfc_slug, branch_name, kind, state, original, proposed, reason,
                 acted_by, acted_at)
            VALUES (?, ?, 'manual', 'accepted', ?, ?, ?, ?, datetime('now'))
            """,
            (
                slug,
                branch,
                prior_body,
                body.new_content,
                f"manual edit: {body.paragraph_count} paragraph(s) changed",
                viewer.user_id,
            ),
        )
        change_id = cur.lastrowid
        try:
            sha = await bot.commit_manual_flush(
                viewer.as_actor(),
                owner=owner,
                repo=repo,
                branch=branch,
                file_path=path,
                new_content=new_file_contents,
                prior_sha=prior_sha,
                change_id=change_id,
                paragraph_count=body.paragraph_count,
                slug=slug,
            )
        except GiteaError as e:
            db.conn().execute("DELETE FROM changes WHERE id = ?", (change_id,))
            raise HTTPException(502, f"Gitea: {e.detail}")

        db.conn().execute(
            "UPDATE changes SET commit_sha = ? WHERE id = ?",
            (sha, change_id),
        )

        # Per §10.6: every manual flush drops a system-author message
        # into the branch chat. Even before the PR exists, the chat is
        # the canonical evidence timeline.
        main_thread_id = _ensure_branch_chat_thread(slug, branch, viewer)
        chat_layer.append_system_message(
            thread_id=main_thread_id,
            text=f"manual edit: {body.paragraph_count} paragraph(s) changed",
        )

        chat_layer.mark_stale_overlapping(rfc_slug=slug, branch_name=branch, new_body=body.new_content)
        await _refresh_cache_for(rfc)

        return {"ok": True, "commit_sha": sha, "change_id": change_id}

    # -------------------------------------------------------------------
    # §17 / §11: visibility + contribute + grants
    # -------------------------------------------------------------------

    @router.post("/api/rfcs/{slug}/branches/{branch:path}/visibility")
    async def set_branch_visibility(slug: str, branch: str, body: VisibilityBody, request: Request) -> dict[str, Any]:
        viewer = auth.require_contributor(request)
        rfc = _require_rfc_with_repo(slug, viewer)
        creator = _branch_creator(slug, branch)
        _require_branch_owner(rfc, viewer, creator)
        current = _branch_vis(slug, branch)
        read_public = body.read_public if body.read_public is not None else current["read_public"]
        contribute_mode = body.contribute_mode or current["contribute_mode"]
        db.conn().execute(
            """
            INSERT INTO branch_visibility (rfc_slug, branch_name, read_public, contribute_mode)
            VALUES (?, ?, ?, ?)
            ON CONFLICT(project_id, rfc_slug, branch_name) DO UPDATE SET
                read_public = excluded.read_public,
                contribute_mode = excluded.contribute_mode
            """,
            (slug, branch, 1 if read_public else 0, contribute_mode),
        )
        return {"ok": True, "visibility": _branch_vis(slug, branch)}

    @router.post("/api/rfcs/{slug}/branches/{branch:path}/grants")
    async def add_branch_grant(slug: str, branch: str, body: GrantBody, request: Request) -> dict[str, Any]:
        viewer = auth.require_contributor(request)
        rfc = _require_rfc_with_repo(slug, viewer)
        creator = _branch_creator(slug, branch)
        _require_branch_owner(rfc, viewer, creator)
        grantee = db.conn().execute(
            "SELECT id FROM users WHERE gitea_login = ?", (body.grantee_gitea_login,)
        ).fetchone()
        if not grantee:
            raise HTTPException(404, f"User '{body.grantee_gitea_login}' has no account in this app")
        db.conn().execute(
            """
            INSERT OR IGNORE INTO branch_contribute_grants
                (rfc_slug, branch_name, grantee_user_id, granted_by)
            VALUES (?, ?, ?, ?)
            """,
            (slug, branch, grantee["id"], viewer.user_id),
        )
        return {"ok": True, "grants": _branch_grants(slug, branch)}

    @router.delete("/api/rfcs/{slug}/branches/{branch:path}/grants/{grantee_login}")
    async def revoke_branch_grant(slug: str, branch: str, grantee_login: str, request: Request) -> dict[str, Any]:
        viewer = auth.require_contributor(request)
        rfc = _require_rfc_with_repo(slug, viewer)
        creator = _branch_creator(slug, branch)
        _require_branch_owner(rfc, viewer, creator)
        grantee = db.conn().execute(
            "SELECT id FROM users WHERE gitea_login = ?", (grantee_login,)
        ).fetchone()
        if grantee:
            db.conn().execute(
                "DELETE FROM branch_contribute_grants WHERE rfc_slug = ? AND branch_name = ? AND grantee_user_id = ?",
                (slug, branch, grantee["id"]),
            )
        return {"ok": True, "grants": _branch_grants(slug, branch)}

    # -------------------------------------------------------------------
    # §17 / §8.12 / §8.13: threads
    # -------------------------------------------------------------------

    @router.get("/api/rfcs/{slug}/branches/{branch:path}/threads")
    async def list_branch_threads(slug: str, branch: str, request: Request) -> dict[str, Any]:
        viewer = auth.current_user(request)
        _require_rfc_with_repo(slug, viewer)
        if not _can_read_branch(slug, branch, viewer):
            raise HTTPException(403, "Branch is private")
        rows = db.conn().execute(
            """
            SELECT id, anchor_kind, anchor_payload, thread_kind, label, state,
                   created_by, created_at, resolved_at, resolved_by
              FROM threads
             WHERE rfc_slug = ? AND branch_name = ?
             ORDER BY id
            """,
            (slug, branch),
        ).fetchall()
        return {"items": [_serialize_thread(r) for r in rows]}

    @router.post("/api/rfcs/{slug}/branches/{branch:path}/threads")
    async def create_branch_thread(slug: str, branch: str, body: ThreadCreateBody, request: Request) -> dict[str, Any]:
        viewer = auth.require_contributor(request)
        _require_rfc_with_repo(slug, viewer)
        if body.thread_kind == "flag" and not body.label:
            raise HTTPException(422, "Flag threads require a label")
        cur = db.conn().execute(
            """
            INSERT INTO threads (rfc_slug, branch_name, anchor_kind, anchor_payload,
                                 thread_kind, label, created_by)
            VALUES (?, ?, ?, ?, ?, ?, ?)
            """,
            (
                slug,
                branch,
                body.anchor_kind,
                json.dumps(body.anchor_payload) if body.anchor_payload else None,
                body.thread_kind,
                body.label,
                viewer.user_id,
            ),
        )
        thread_id = cur.lastrowid
        message_id = None
        if body.message and body.thread_kind == "chat":
            message_id = chat_layer.append_user_message(
                thread_id=thread_id, author_user_id=viewer.user_id, text=body.message, quote=None
            )
        return {"thread_id": thread_id, "message_id": message_id}

    @router.get("/api/rfcs/{slug}/branches/{branch:path}/threads/{thread_id}/messages")
    async def get_thread_messages(slug: str, branch: str, thread_id: int, request: Request) -> dict[str, Any]:
        viewer = auth.current_user(request)
        _require_rfc_with_repo(slug, viewer)
        if not _can_read_branch(slug, branch, viewer):
            raise HTTPException(403, "Branch is private")
        thread = _require_thread(slug, branch, thread_id)
        rows = db.conn().execute(
            """
            SELECT m.id, m.role, m.author_user_id, u.gitea_login as author_login,
                   u.display_name as author_display, m.model_id, m.text, m.quote, m.created_at
              FROM thread_messages m
         LEFT JOIN users u ON u.id = m.author_user_id
             WHERE m.thread_id = ?
             ORDER BY m.id
            """,
            (thread_id,),
        ).fetchall()
        return {
            "thread": _serialize_thread(thread),
            "messages": [_serialize_message(r) for r in rows],
        }

    @router.post("/api/rfcs/{slug}/branches/{branch:path}/threads/{thread_id}/messages")
    async def post_thread_message(
        slug: str, branch: str, thread_id: int, body: ThreadMessageBody, request: Request
    ) -> dict[str, Any]:
        viewer = auth.require_contributor(request)
        _require_rfc_with_repo(slug, viewer)
        _require_thread(slug, branch, thread_id)
        if not _can_read_branch(slug, branch, viewer):
            raise HTTPException(403, "Branch is private")
        message_id = chat_layer.append_user_message(
            thread_id=thread_id,
            author_user_id=viewer.user_id,
            text=body.text,
            quote=body.quote,
        )
        return {"ok": True, "message_id": message_id}

    @router.post("/api/rfcs/{slug}/branches/{branch:path}/chat-seen")
    async def advance_chat_seen(slug: str, branch: str, body: dict, request: Request) -> dict[str, Any]:
        """§15.7 chat-seen cursor advance.

        Body: `{"last_seen_message_id": <int>}`. Upserts branch_chat_seen
        and runs the §15.7 reconciler — every unread notification scoped
        to this (slug, branch) on or before the new cursor is marked read.
        """
        viewer = auth.require_user(request)
        _require_rfc_with_repo(slug, viewer)
        if not _can_read_branch(slug, branch, viewer):
            raise HTTPException(403, "Branch is private")
        last_seen = int(body.get("last_seen_message_id") or 0) or None
        db.conn().execute(
            """
            INSERT INTO branch_chat_seen (user_id, rfc_slug, branch_name, last_seen_message_id, seen_at)
            VALUES (?, ?, ?, ?, datetime('now'))
            ON CONFLICT(project_id, user_id, rfc_slug, branch_name) DO UPDATE SET
                last_seen_message_id = excluded.last_seen_message_id,
                seen_at = excluded.seen_at
            """,
            (viewer.user_id, slug, branch, last_seen),
        )
        from . import notify
        reconciled = notify.reconcile_seen_advance(
            user_id=viewer.user_id, rfc_slug=slug, branch_name=branch,
        )
        return {"ok": True, "reconciled": reconciled}

    @router.post("/api/rfcs/{slug}/branches/{branch:path}/threads/{thread_id}/resolve")
    async def resolve_thread(slug: str, branch: str, thread_id: int, request: Request) -> dict[str, Any]:
        viewer = auth.require_contributor(request)
        rfc = _require_rfc_with_repo(slug, viewer)
        thread = _require_thread(slug, branch, thread_id)
        creator = _branch_creator(slug, branch)
        if not _can_resolve_thread(rfc, thread, creator, viewer):
            raise HTTPException(403, "Only the thread creator, the branch creator, an RFC owner/arbiter, or an app admin/owner may resolve")
        db.conn().execute(
            """
            UPDATE threads SET state = 'resolved', resolved_by = ?, resolved_at = datetime('now')
             WHERE id = ?
            """,
            (viewer.user_id, thread_id),
        )
        return {"ok": True, "thread_id": thread_id}

    # -------------------------------------------------------------------
    # §17 / §18 carryover: SSE-streaming chat turn on a thread
    # -------------------------------------------------------------------

    @router.post("/api/rfcs/{slug}/branches/{branch:path}/threads/{thread_id}/chat")
    async def stream_chat_turn(
        slug: str, branch: str, thread_id: int, body: ChatTurnBody, request: Request
    ):
        viewer = auth.require_contributor(request)
        rfc = _require_rfc_with_repo(slug, viewer)
        thread = _require_thread(slug, branch, thread_id)
        if not _can_read_branch(slug, branch, viewer):
            raise HTTPException(403, "Branch is private")
        # §6.6: option list and default come from the per-RFC resolved set,
        # not the operator universe. body.model is honored only if it sits
        # inside the resolved set; otherwise we fall to the RFC default.
        resolved = models_resolver.resolve_models_for_rfc(slug, providers)
        if not resolved:
            raise HTTPException(503, "No AI providers configured")
        model_key = body.model if body.model in resolved else resolved[0]
        # §6.7: pick the funder-credentialed instance when a consenting
        # funder is in effect on this RFC; otherwise the operator's.
        provider = funder.provider_for_rfc(slug, model_key, providers)
        if provider is None:
            raise HTTPException(503, "No AI providers configured")

        # Fetch the live branch body so the prompt is anchored to
        # what's in Gitea right now, not the cache. For super-draft,
        # extract just the body part from the entry envelope.
        owner, repo = _repo_for(rfc, branch)
        path = _file_path_for(rfc, branch)
        fetched = await gitea.read_file(owner, repo, path, ref=branch)
        body_text = _extract_body(rfc, fetched[0], branch) if fetched else ""

        prompt_text = body.text
        if body.quote:
            prompt_text = f'The contributor has selected this passage:\n"{body.quote}"\n\n---\n\n{body.text}'

        user_message_id = chat_layer.append_user_message(
            thread_id=thread_id, author_user_id=viewer.user_id, text=body.text, quote=body.quote
        )
        assistant_message_id = chat_layer.append_assistant_placeholder(
            thread_id=thread_id, model_id=model_key
        )

        system = chat_layer.build_system_prompt(title=rfc["title"], body=body_text)
        rows = db.conn().execute(
            """
            SELECT role, text FROM thread_messages
             WHERE thread_id = ? AND id < ? AND role IN ('user', 'assistant')
             ORDER BY id
            """,
            (thread_id, user_message_id),
        ).fetchall()
        history = [{"role": r["role"], "content": r["text"]} for r in rows]

        async def event_stream():
            async for chunk in chat_layer.stream_assistant_turn(
                provider=provider,
                system_prompt=system,
                history=history,
                user_message=prompt_text,
                thread_id=thread_id,
                rfc_slug=slug,
                branch_name=branch,
                assistant_message_id=assistant_message_id,
            ):
                yield chunk

        headers = {
            "X-Assistant-Message-Id": str(assistant_message_id),
            "X-User-Message-Id": str(user_message_id),
            "Cache-Control": "no-cache",
        }
        return StreamingResponse(event_stream(), media_type="text/event-stream", headers=headers)

    # -------------------------------------------------------------------
    # §17: GET /api/rfcs/<slug>/branches/<branch>
    # Per §4: branch bodies are NOT cached — fetch live from Gitea.
    # Per §9.5 / §17: when slug resolves to super-draft, <branch> names
    # a meta-repo branch and the underlying file is rfcs/<slug>.md with
    # the body wrapped in frontmatter.
    #
    # Declared LAST among the branch-scoped GET routes per the §19.2
    # "branch-name path routing" candidate: `{branch:path}` is greedy
    # by Starlette's converter, so `branches/foo/threads` would match
    # this bare GET with branch=foo/threads if declared first. Putting
    # the more-specific `threads` and `threads/{thread_id}/messages`
    # GETs above lets them claim those URLs; this one catches anything
    # else, including slashed branch names like `foo/bar`.
    # -------------------------------------------------------------------

    @router.get("/api/rfcs/{slug}/branches/{branch:path}")
    async def get_branch_view(slug: str, branch: str, request: Request) -> dict[str, Any]:
        viewer = auth.current_user(request)
        rfc = _require_rfc_with_repo(slug, viewer)
        if not _can_read_branch(slug, branch, viewer):
            raise HTTPException(403, "Branch is private")

        owner, repo = _repo_for(rfc, branch)
        path = _file_path_for(rfc, branch)
        result = await gitea.read_file(owner, repo, path, ref=branch)
        if result is None:
            br = await gitea.get_branch(owner, repo, branch)
            if br is None:
                raise HTTPException(404, "Branch not found")
            body, body_sha = "", ""
        else:
            content, body_sha = result
            body = _extract_body(rfc, content, branch)

        # Ensure the whole-doc chat thread for the branch exists.
        thread_id = _ensure_branch_chat_thread(slug, branch, viewer)

        # Sub-threads (range/paragraph) and flags scoped to this branch.
        thread_rows = db.conn().execute(
            """
            SELECT id, anchor_kind, anchor_payload, thread_kind, label, state, created_by, created_at
              FROM threads
             WHERE rfc_slug = ? AND branch_name = ?
             ORDER BY id
            """,
            (slug, branch),
        ).fetchall()
        threads = [_serialize_thread(t) for t in thread_rows]

        # Visibility, contribute, grants.
        vis = _branch_vis(slug, branch)
        grants = _branch_grants(slug, branch)

        # Pending and resolved changes scoped to this branch.
        changes_rows = db.conn().execute(
            """
            SELECT id, thread_id, source_message_id, kind, state, original, proposed, reason,
                   was_edited_before_accept, stale_since, acted_by, acted_at, commit_sha, created_at
              FROM changes
             WHERE rfc_slug = ? AND branch_name = ?
             ORDER BY id
            """,
            (slug, branch),
        ).fetchall()
        changes = [_serialize_change(c) for c in changes_rows]

        # Branch metadata for the breadcrumb / header.
        creator = _branch_creator(slug, branch)
        capabilities = _capabilities(rfc, slug, branch, viewer, creator)

        return {
            "slug": slug,
            "title": rfc["title"],
            "branch_name": branch,
            "body": body,
            "body_sha": body_sha,
            "main_thread_id": thread_id,
            "threads": threads,
            "changes": changes,
            "visibility": vis,
            "grants": grants,
            "creator": creator,
            "capabilities": capabilities,
        }

    # ------------------------------------------------------------------
    # Permission + state helpers (closures, share `config` etc.)
    # ------------------------------------------------------------------

    def _require_rfc(slug: str, viewer):
        row = db.conn().execute("SELECT * FROM cached_rfcs WHERE slug = ?", (slug,)).fetchone()
        if row is None:
            raise HTTPException(404, "RFC not found")
        # §22.5 visibility gate (subtractive, §22.7): a gated project's entries
        # 404 to non-members — indistinguishable from an unknown slug.
        auth.require_project_readable(viewer, row["project_id"])
        return row

    def _require_rfc_with_repo(slug: str, viewer):
        """Used by every branch-scoped endpoint. Under the meta-only
        topology (§1) the meta repo is the implicit target for every
        entry — super-draft and active alike — so there is no per-RFC
        repo check. The name is retained for call-site stability; a
        withdrawn entry is still rejected."""
        row = _require_rfc(slug, viewer)
        if row["state"] == "withdrawn":
            raise HTTPException(409, "RFC is withdrawn")
        return row

    def _require_active_rfc(slug: str, viewer):
        row = _require_rfc_with_repo(slug, viewer)
        if row["state"] != "active":
            raise HTTPException(409, f"RFC is {row['state']}, not active")
        return row

    def _require_super_draft(slug: str, viewer):
        row = _require_rfc(slug, viewer)
        if row["state"] != "super-draft":
            raise HTTPException(409, f"RFC is {row['state']}, not super-draft")
        return row

    def _is_super_draft(rfc) -> bool:
        return rfc["state"] == "super-draft"

    def _is_meta_resident(rfc) -> bool:
        """Meta-only topology (§1): an entry lives in the meta repo's
        `rfcs/<slug>.md` (super-draft or active-in-place) unless it carries
        a legacy per-RFC `repo:` — which nothing does after the RFC-0001
        fold-back (§13.6)."""
        return not rfc["repo"]

    def _is_meta_branch_name(name: str) -> bool:
        """A branch name shaped like one of the bot's meta-repo prefixes.
        Retained for the legacy per-RFC-repo read path; under meta-only
        every entry is already a meta target via `_is_meta_resident`."""
        return name != "main" and name.startswith((
            "edit-", "edit/", "metadata-", "metadata/", "claim/", "propose/",
            "graduate-",
        ))

    def _is_meta_target(rfc, branch: str) -> bool:
        """A meta-resident entry (super-draft or active-in-place, §1)
        targets the meta repo for every branch. The branch-name fallback
        covers the retired per-RFC-repo case for any legacy entry that
        still carries a `repo:`."""
        if _is_meta_resident(rfc):
            return True
        return _is_meta_branch_name(branch)

    def _repo_for(rfc, branch: str = "main") -> tuple[str, str]:
        if _is_meta_target(rfc, branch):
            return config.gitea_org, (projects_mod.default_content_repo(config) or "")
        owner, repo = rfc["repo"].split("/", 1)
        return owner, repo

    def _file_path_for(rfc, branch: str = "main") -> str:
        if _is_meta_target(rfc, branch):
            return f"rfcs/{rfc['slug']}.md"
        return RFC_FILE_PATH

    def _extract_body(rfc, file_contents: str, branch: str = "main") -> str:
        """For super-draft entries (and active-RFC pre-graduation reads
        per §9.8) the file on disk is the full frontmatter+body envelope;
        the editable body is entry.body. For active RFCs reading their
        per-RFC repo the file is just RFC.md and the whole thing is body."""
        if not _is_meta_target(rfc, branch):
            return file_contents
        try:
            entry = entry_mod.parse(file_contents)
        except Exception:
            return file_contents
        return entry.body

    def _wrap_body(rfc, prior_contents: str, new_body: str, branch: str = "main") -> str:
        """Inverse of _extract_body: re-wrap a new body into the entry
        envelope, preserving the prior frontmatter exactly."""
        if not _is_meta_target(rfc, branch):
            return new_body
        entry = entry_mod.parse(prior_contents)
        # Ensure exactly one trailing newline so the serializer's
        # round-trip is stable.
        entry.body = new_body if new_body.endswith("\n") else new_body + "\n"
        return entry_mod.serialize(entry)

    async def _refresh_cache_for(rfc) -> None:
        if _is_meta_resident(rfc):
            await cache.refresh_meta_repo(config, gitea)
            await cache.refresh_meta_branches(config, gitea)
        else:
            await cache.refresh_rfc_repo(config, gitea, rfc["slug"])

    def _ensure_branch_chat_thread(slug: str, branch: str, viewer) -> int:
        """Per §8.12: every branch has a default whole-doc chat thread.
        Create it lazily on first read. The created_by is null when an
        anonymous viewer triggers creation — the thread is structurally
        owned by the branch, not by whoever opened the view."""
        row = db.conn().execute(
            """
            SELECT id FROM threads
             WHERE rfc_slug = ? AND branch_name = ?
               AND anchor_kind = 'whole-doc' AND thread_kind = 'chat'
             ORDER BY id LIMIT 1
            """,
            (slug, branch),
        ).fetchone()
        if row:
            return row["id"]
        cur = db.conn().execute(
            """
            INSERT INTO threads (rfc_slug, branch_name, anchor_kind, thread_kind, label, created_by)
            VALUES (?, ?, 'whole-doc', 'chat', NULL, ?)
            """,
            (slug, branch, viewer.user_id if viewer else None),
        )
        return cur.lastrowid

    def _ensure_branch_vis(slug: str, branch: str, *, creator_user_id: int) -> None:
        del creator_user_id  # creator is sourced from actions log
        db.conn().execute(
            """
            INSERT OR IGNORE INTO branch_visibility (rfc_slug, branch_name, read_public, contribute_mode)
            VALUES (?, ?, 1, 'just-me')
            """,
            (slug, branch),
        )

    def _branch_vis(slug: str, branch: str) -> dict:
        row = db.conn().execute(
            "SELECT read_public, contribute_mode FROM branch_visibility WHERE rfc_slug = ? AND branch_name = ?",
            (slug, branch),
        ).fetchone()
        if row:
            return {"read_public": bool(row["read_public"]), "contribute_mode": row["contribute_mode"]}
        return {"read_public": True, "contribute_mode": "just-me"}

    def _branch_grants(slug: str, branch: str) -> list[dict]:
        rows = db.conn().execute(
            """
            SELECT u.gitea_login, u.display_name, g.granted_at
              FROM branch_contribute_grants g
              JOIN users u ON u.id = g.grantee_user_id
             WHERE g.rfc_slug = ? AND g.branch_name = ?
             ORDER BY g.granted_at
            """,
            (slug, branch),
        ).fetchall()
        return [{"gitea_login": r["gitea_login"], "display_name": r["display_name"], "granted_at": r["granted_at"]} for r in rows]

    def _branch_creator(slug: str, branch: str) -> str | None:
        if branch == "main":
            return None
        row = db.conn().execute(
            """
            SELECT on_behalf_of FROM actions
             WHERE action_kind = 'create_branch' AND rfc_slug = ? AND branch_name = ?
             ORDER BY id LIMIT 1
            """,
            (slug, branch),
        ).fetchone()
        return row["on_behalf_of"] if row else None

    def _can_read_branch(slug: str, branch: str, viewer) -> bool:
        # §22.5 visibility gate first (subtractive, §22.7): in a gated project
        # nothing — not even main or a read_public branch — is readable by a
        # non-member.
        pid = auth.project_of_rfc(slug)
        if not auth.can_read_project(viewer, pid):
            return False
        if branch == "main":
            return True
        vis = _branch_vis(slug, branch)
        if vis["read_public"]:
            return True
        if viewer is None:
            return False
        if auth.is_project_superuser(viewer, pid):
            return True
        creator = _branch_creator(slug, branch)
        if creator and viewer.gitea_login == creator:
            return True
        rfc = db.conn().execute("SELECT owners_json, arbiters_json FROM cached_rfcs WHERE slug = ?", (slug,)).fetchone()
        if rfc:
            owners = json.loads(rfc["owners_json"] or "[]")
            arbiters = json.loads(rfc["arbiters_json"] or "[]")
            if viewer.gitea_login in owners or viewer.gitea_login in arbiters:
                return True
        row = db.conn().execute(
            """
            SELECT 1 FROM branch_contribute_grants g
             WHERE g.rfc_slug = ? AND g.branch_name = ? AND g.grantee_user_id = ?
            """,
            (slug, branch, viewer.user_id),
        ).fetchone()
        return row is not None

    def _can_contribute(rfc, slug: str, branch: str, viewer) -> bool:
        if viewer is None:
            return False
        if branch == "main":
            return False
        # §9.8 (LEGACY per-repo only): pre-graduation history branches are
        # read-only on the post-graduation surface of a per-RFC-repo active
        # entry. Under the meta-only topology (§1, §13.4) an active RFC's
        # `edit-<slug>-…` branches are its *current* editable branches, not
        # a frozen pre-graduation set, so this guard applies only when a
        # legacy `repo:` is set (nothing, after the RFC-0001 fold-back).
        if rfc["state"] == "active" and rfc["repo"] and _is_meta_branch_name(branch):
            return False
        pid = auth.project_of_rfc(slug)
        # §22.5 visibility gate (subtractive): no contribute in an unreadable
        # project.
        if not auth.can_read_project(viewer, pid):
            return False
        if auth.is_project_superuser(viewer, pid):
            return True
        owners = json.loads(rfc["owners_json"] or "[]")
        arbiters = json.loads(rfc["arbiters_json"] or "[]")
        if viewer.gitea_login in owners or viewer.gitea_login in arbiters:
            return True
        creator = _branch_creator(slug, branch)
        if creator and viewer.gitea_login == creator:
            return True
        vis = _branch_vis(slug, branch)
        if vis["contribute_mode"] == "any-contributor":
            # "any contributor" means anyone with project-level write standing
            # (§22.6/§22.7) — the implicit-public baseline on a public project,
            # or an explicit project_contributor/admin elsewhere.
            return auth.can_contribute_in_project(viewer, pid)
        if vis["contribute_mode"] == "specific":
            row = db.conn().execute(
                """
                SELECT 1 FROM branch_contribute_grants
                 WHERE rfc_slug = ? AND branch_name = ? AND grantee_user_id = ?
                """,
                (slug, branch, viewer.user_id),
            ).fetchone()
            return row is not None
        return False

    def _require_can_contribute(slug: str, branch: str, viewer) -> None:
        rfc = db.conn().execute(
            "SELECT state, repo, owners_json, arbiters_json FROM cached_rfcs WHERE slug = ?",
            (slug,),
        ).fetchone()
        if not _can_contribute(rfc, slug, branch, viewer):
            raise HTTPException(403, "You do not have contribute access to this branch")

    def _require_branch_owner(rfc, viewer, creator: str | None) -> None:
        # §22.6: a project_admin is the per-RFC owner/arbiter authority lifted
        # to project scope, so it (and a deployment owner/admin) clears here.
        if auth.is_project_superuser(viewer, rfc["project_id"]):
            return
        owners = json.loads(rfc["owners_json"] or "[]")
        arbiters = json.loads(rfc["arbiters_json"] or "[]")
        if viewer.gitea_login in owners or viewer.gitea_login in arbiters:
            return
        if creator and viewer.gitea_login == creator:
            return
        raise HTTPException(403, "Only the branch creator, an RFC owner/arbiter, or an admin/owner may change branch settings")

    def _can_edit_metadata(rfc, viewer) -> bool:
        """§9.5: super-draft owners/arbiters per §6.3 plus project_admin /
        app admins/owners (§22.6). Until §13.1's claim runs, the super-draft
        has no owners, so the set collapses to the superuser tier only —
        sensible because admin oversight is the only path to canonicalizing
        edits on an unclaimed entry."""
        if auth.is_project_superuser(viewer, rfc["project_id"]):
            return True
        owners = json.loads(rfc["owners_json"] or "[]")
        arbiters = json.loads(rfc["arbiters_json"] or "[]")
        return viewer.gitea_login in owners or viewer.gitea_login in arbiters

    def _capabilities(rfc, slug: str, branch: str, viewer, creator: str | None) -> dict:
        owners = json.loads(rfc["owners_json"] or "[]")
        arbiters = json.loads(rfc["arbiters_json"] or "[]")
        return {
            "can_read": _can_read_branch(slug, branch, viewer),
            "can_contribute": _can_contribute(rfc, slug, branch, viewer) if viewer else False,
            "can_change_branch_settings": viewer is not None and (
                auth.is_project_superuser(viewer, rfc["project_id"])
                or (creator is not None and viewer.gitea_login == creator)
                or viewer.gitea_login in (owners + arbiters)
            ),
            "can_edit_metadata": viewer is not None and _is_super_draft(rfc) and _can_edit_metadata(rfc, viewer),
            "is_anonymous": viewer is None,
        }

    def _branch_summary(slug: str, br, viewer) -> dict:
        return {
            "name": br["branch_name"],
            "head_sha": br["head_sha"],
            "state": br["state"],
            "last_commit_at": br["last_commit_at"],
            "pinned": bool(br["pinned"]),
            "creator": _branch_creator(slug, br["branch_name"]),
            "visibility": _branch_vis(slug, br["branch_name"]),
        }

    def _require_change(slug: str, branch: str, change_id: int):
        row = db.conn().execute(
            "SELECT * FROM changes WHERE id = ? AND rfc_slug = ? AND branch_name = ?",
            (change_id, slug, branch),
        ).fetchone()
        if not row:
            raise HTTPException(404, "Change not found on this branch")
        return row

    def _require_pending_change(slug: str, branch: str, change_id: int):
        row = _require_change(slug, branch, change_id)
        if row["state"] != "pending":
            raise HTTPException(409, f"Change is already {row['state']}")
        return row

    def _require_thread(slug: str, branch: str, thread_id: int):
        row = db.conn().execute(
            "SELECT * FROM threads WHERE id = ? AND rfc_slug = ? AND branch_name = ?",
            (thread_id, slug, branch),
        ).fetchone()
        if not row:
            raise HTTPException(404, "Thread not found on this branch")
        return row

    def _can_resolve_thread(rfc, thread, creator: str | None, viewer) -> bool:
        if viewer is None:
            return False
        if auth.is_project_superuser(viewer, rfc["project_id"]):
            return True
        owners = json.loads(rfc["owners_json"] or "[]")
        arbiters = json.loads(rfc["arbiters_json"] or "[]")
        if viewer.gitea_login in owners or viewer.gitea_login in arbiters:
            return True
        if creator and viewer.gitea_login == creator:
            return True
        if thread["created_by"] == viewer.user_id:
            return True
        return False

    return router


# ---------------------------------------------------------------------------
# Serialization helpers (module-level for clarity)
# ---------------------------------------------------------------------------


def _serialize_thread(row) -> dict[str, Any]:
    payload = row["anchor_payload"]
    try:
        anchor = json.loads(payload) if payload else None
    except Exception:
        anchor = None
    return {
        "id": row["id"],
        "anchor_kind": row["anchor_kind"],
        "anchor_payload": anchor,
        "thread_kind": row["thread_kind"],
        "label": row["label"],
        "state": row["state"],
        "created_by": row["created_by"],
        "created_at": row["created_at"],
        "resolved_at": row["resolved_at"] if "resolved_at" in row.keys() else None,
        "resolved_by": row["resolved_by"] if "resolved_by" in row.keys() else None,
    }


def _serialize_change(row) -> dict[str, Any]:
    return {
        "id": row["id"],
        "thread_id": row["thread_id"],
        "source_message_id": row["source_message_id"],
        "kind": row["kind"],
        "state": row["state"],
        "original": row["original"],
        "proposed": row["proposed"],
        "reason": row["reason"],
        "was_edited_before_accept": bool(row["was_edited_before_accept"]),
        "stale_since": row["stale_since"],
        "acted_by": row["acted_by"],
        "acted_at": row["acted_at"],
        "commit_sha": row["commit_sha"],
        "created_at": row["created_at"],
    }


def _serialize_message(row) -> dict[str, Any]:
    return {
        "id": row["id"],
        "role": row["role"],
        "author_user_id": row["author_user_id"],
        "author_login": row["author_login"],
        "author_display": row["author_display"],
        "model_id": row["model_id"],
        "text": row["text"],
        "quote": row["quote"],
        "created_at": row["created_at"],
    }


# ---------------------------------------------------------------------------
# Branch name validation + auto-generation per §8.14 / §9.5
# ---------------------------------------------------------------------------

_BRANCH_NAME_RE = re.compile(r"^[a-z0-9][a-z0-9._\-/]*[a-z0-9]$")


def _validate_branch_name(name: str) -> None:
    # §8.14: "exact format is an implementation detail." We accept Git's
    # standard ref-friendly subset and reject anything with whitespace,
    # leading/trailing punctuation, or path components Gitea would refuse.
    if len(name) > 120 or not _BRANCH_NAME_RE.match(name):
        raise HTTPException(422, "Branch name must be lowercase alphanumerics, hyphens, dots, slashes")
    # Reserved-prefix guard: these are bot-internal naming conventions.
    # Slice 4 added `edit-` and `metadata-` to dodge the §19.2 path-
    # routing candidate while keeping the §9.5 structural shape legible.
    if name == "main" or name.startswith(("propose/", "edit/", "edit-", "claim/", "metadata/", "metadata-")):
        raise HTTPException(422, "Branch name conflicts with a reserved prefix")


def _auto_branch_name(login: str) -> str:
    # Per §8.14: auto-generated value, exact format implementation detail.
    # `<login>-draft-<hex>` keeps the branch's origin legible in the Git
    # log and avoids slashes per the §19.2 path-routing candidate.
    import secrets
    return f"{login.lower()}-draft-{secrets.token_hex(3)}"


def _auto_edit_branch_name(slug: str) -> str:
    # Per §9.5: structural form is `edit/<slug>/<auto-name>`; Slice 4
    # uses `edit-<slug>-<6hex>` to dodge the §19.2 path-routing
    # candidate — three components separated by dashes, with the slug as
    # the second component so the cache parsers can recover it.
    import secrets
    return f"edit-{slug}-{secrets.token_hex(3)}"