rfc-app/backend/app/api_graduation.py

"""§13 graduation flow — the meta-only in-place state flip.

Under the meta-only topology (SPEC §1), graduation no longer creates a
per-RFC repo. It is a single frontmatter-flipping commit to the entry's
`rfcs/<slug>.md` on the meta repo: open a PR that re-serializes the entry
with `state: active`, the assigned integer `id`, `graduated_at` /
`graduated_by`, and the dialog's owners — **leaving the body unchanged** —
then auto-merge it. There is no repo to create, nothing to seed, and the
body is neither moved nor stripped, so there is no multi-step transaction
and no rollback (§13.3). If the open or merge fails, the entry stays a
super-draft and we clean up the half-open PR/branch (the only artifact a
mid-flip failure can leave behind).

Routes (§17):

  - GET  /api/rfcs/<slug>/graduate/check       (§13.2 debounced validator)
  - POST /api/rfcs/<slug>/graduate             (§13.3 the flip)
  - GET  /api/rfcs/<slug>/graduate/progress    (§13.3 SSE step stream)
  - GET  /api/rfcs/<slug>/blocking-prs         (informational; no longer a
                                                graduation precondition)

Plus the §13.1 claim PR endpoint (POST /api/rfcs/<slug>/claim) and the
§13.7 retire / un-retire endpoints (POST /api/rfcs/<slug>/retire,
POST /api/rfcs/<slug>/unretire) — soft-delete and its site-owner-only
reversal, each a single in-place frontmatter flip via an auto-merged PR.

The orchestrator runs in-process — each in-flight graduation lives in a
small `GraduationState` keyed by slug, with an asyncio.Queue feeding the
SSE handler. §13.4's chat/branch/history are a database no-op: every row
is keyed by the slug per §2.3 and stays put across the flip.
"""
from __future__ import annotations

import asyncio
import json
import logging
import re
from dataclasses import dataclass, field
from typing import Any

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

from . import auth, cache, db, entry as entry_mod, metadata as metadata_mod, projects as projects_mod
from .bot import Actor, Bot
from .config import Config
from .gitea import Gitea, GiteaError

log = logging.getLogger(__name__)


# ---------------------------------------------------------------------------
# Step machine — two steps under meta-only: open the flip PR, merge it.
# ---------------------------------------------------------------------------


STEP_KEYS = (
    "open_pr",
    "merge_pr",
)

STEP_LABELS = {
    "open_pr": "Open graduation PR (flip state to active)",
    "merge_pr": "Merge graduation PR",
}


@dataclass
class StepState:
    key: str
    label: str
    status: str = "pending"  # pending|running|done|failed|not-reached
    detail: str = ""


@dataclass
class GraduationState:
    slug: str
    rfc_id: str | None  # null when graduating without a number (§13.2)
    owners: list[str]
    arbiters: list[str]
    steps: list[StepState]
    queue: asyncio.Queue = field(default_factory=asyncio.Queue)
    finished: bool = False
    succeeded: bool = False
    error: str | None = None
    new_pr_number: int | None = None
    graduation_branch: str | None = None

    def to_payload(self) -> dict:
        return {
            "slug": self.slug,
            "rfc_id": self.rfc_id,
            "steps": [_step_payload(s) for s in self.steps],
            "finished": self.finished,
            "succeeded": self.succeeded,
            "error": self.error,
            "pr_number": self.new_pr_number,
        }


def _step_payload(s: StepState) -> dict:
    return {"key": s.key, "label": s.label, "status": s.status, "detail": s.detail}


# Process-local registry. Single-process FastAPI per §4.2 means in-memory
# is fine; the registry is keyed by slug to refuse concurrent graduations
# of the same entry (the §13.2 atomic re-check is a separate defense
# against a concurrent attempt of a DIFFERENT slug claiming the same
# integer ID).
_active: dict[str, GraduationState] = {}


def _get_active(slug: str) -> GraduationState | None:
    return _active.get(slug)


def _new_active(slug: str, *, rfc_id: str | None,
                owners: list[str], arbiters: list[str]) -> GraduationState:
    state = GraduationState(
        slug=slug, rfc_id=rfc_id,
        owners=owners, arbiters=arbiters,
        steps=[StepState(key=k, label=STEP_LABELS[k]) for k in STEP_KEYS],
    )
    _active[slug] = state
    return state


# ---------------------------------------------------------------------------
# Validation helpers
# ---------------------------------------------------------------------------


_RFC_ID_RE = re.compile(r"^RFC-\d{4,}$")


def _is_valid_rfc_id(rfc_id: str) -> bool:
    return bool(_RFC_ID_RE.match(rfc_id))


def _suggest_next_rfc_id() -> str:
    rows = db.conn().execute(
        "SELECT rfc_id FROM cached_rfcs WHERE rfc_id LIKE 'RFC-%'"
    ).fetchall()
    used: set[int] = set()
    for r in rows:
        try:
            used.add(int(r["rfc_id"].split("-", 1)[1]))
        except (IndexError, ValueError):
            continue
    nxt = (max(used) + 1) if used else 1
    return f"RFC-{nxt:04d}"


def _rfc_id_taken(rfc_id: str, *, excluding_slug: str) -> bool:
    row = db.conn().execute(
        "SELECT slug FROM cached_rfcs WHERE rfc_id = ? AND slug != ?",
        (rfc_id, excluding_slug),
    ).fetchone()
    return row is not None


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


class GraduateBody(BaseModel):
    # §13.2: the integer id is OPTIONAL. Blank/absent → graduate with no
    # number (id stays null, the slug is the canonical identifier per
    # §2.3). When supplied it must match ^RFC-\d{4,}$ and be free — both
    # are checked in the handler, not by the field bound (an empty string
    # is a legitimate value here).
    rfc_id: str | None = Field(default=None, max_length=40)
    owners: list[str] = Field(min_length=1)


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


def make_router(
    config: Config,
    gitea: Gitea,
    bot: Bot,
) -> APIRouter:
    router = APIRouter()

    # -------------------------------------------------------------------
    # GET /api/rfcs/<slug>/blocking-prs
    # Lists open meta-repo body-edit PRs against rfcs/<slug>.md. Under the
    # meta-only topology (§9.8) these no longer block graduation — the body
    # is kept, so a body-edit PR coexists with the flip. Retained as an
    # informational surface (the dialog can show "N body-edit PRs open").
    # -------------------------------------------------------------------

    @router.get("/api/rfcs/{slug}/blocking-prs")
    async def list_blocking_prs(slug: str, request: Request) -> dict[str, Any]:
        viewer = auth.current_user(request)
        rfc = _require_super_draft(slug, viewer)
        rows = db.conn().execute(
            """
            SELECT pr_number, title, opened_by, opened_at, head_branch, pr_kind
              FROM cached_prs
             WHERE rfc_slug = ?
               AND state = 'open'
               AND pr_kind = 'meta_body_edit'
             ORDER BY opened_at DESC
            """,
            (slug,),
        ).fetchall()
        owners = json.loads(rfc["owners_json"] or "[]")
        arbiters = json.loads(rfc["arbiters_json"] or "[]")
        items = []
        for r in rows:
            can_merge = (
                viewer is not None
                and (
                    auth.is_collection_superuser(viewer, rfc["collection_id"])
                    or viewer.gitea_login in owners
                    or viewer.gitea_login in arbiters
                )
            )
            can_withdraw = (
                viewer is not None
                and (
                    can_merge
                    or viewer.gitea_login == (r["opened_by"] or "")
                )
            )
            items.append({
                "pr_number": r["pr_number"],
                "title": r["title"],
                "author": r["opened_by"],
                "last_activity_at": r["opened_at"],
                "head_branch": r["head_branch"],
                "actions": {
                    "merge": can_merge,
                    "withdraw": can_withdraw,
                    "open_in_new_tab": True,
                },
            })
        # `blocking` is a legacy field name kept for client compatibility;
        # under §9.8 these PRs do not block graduation.
        return {"items": items, "blocking": False}

    # -------------------------------------------------------------------
    # GET /api/rfcs/<slug>/graduate/check?id=
    # Inline validation for the Graduate dialog — debounced from the
    # client. Two fields under meta-only: the integer ID and the owners
    # precondition. There is no repo name to validate (§13.2).
    # -------------------------------------------------------------------

    @router.get("/api/rfcs/{slug}/graduate/check")
    async def graduate_check(
        slug: str, request: Request,
    ) -> dict[str, Any]:
        viewer = auth.current_user(request)
        rfc = _require_super_draft(slug, viewer)
        del viewer  # no permission gate — the dialog only shows up for
                    # admins/owners, but the check itself is read-only.

        candidate_id = (request.query_params.get("id") or "").strip()
        owners = json.loads(rfc["owners_json"] or "[]")

        # ID field — §13.2: the integer id is OPTIONAL. A blank id is
        # valid and means "graduate without a number" (id stays null, the
        # slug is canonical per §2.3). Only a *non-blank* id is held to the
        # RFC-NNNN regex and the collision check.
        id_payload: dict[str, Any] = {"value": candidate_id, "ok": True, "error": None}
        if not candidate_id:
            pass  # blank is the default-accepted case, not an error
        elif not _is_valid_rfc_id(candidate_id):
            id_payload["ok"] = False
            id_payload["error"] = (
                "ID must look like RFC-NNNN (at least four digits), "
                "or leave blank to graduate without a number"
            )
        elif _rfc_id_taken(candidate_id, excluding_slug=slug):
            id_payload["ok"] = False
            id_payload["error"] = f"Integer ID {candidate_id} is already taken"

        # Owners precondition — §13's opening paragraph.
        owners_payload: dict[str, Any] = {
            "ok": len(owners) > 0,
            "count": len(owners),
            "current": owners,
            "error": None if len(owners) > 0 else "No owners claimed yet",
        }

        in_flight = _get_active(slug)
        any_invalid = not (id_payload["ok"] and owners_payload["ok"])

        return {
            "slug": slug,
            "id": id_payload,
            "owners": owners_payload,
            "can_submit": (not any_invalid) and (in_flight is None or in_flight.finished),
            "in_flight": (
                None if in_flight is None
                else {"finished": in_flight.finished, "succeeded": in_flight.succeeded}
            ),
        }

    # -------------------------------------------------------------------
    # POST /api/rfcs/<slug>/graduate
    # Atomic re-validation, then kicks off the flip as an async task.
    # The client opens GET /graduate/progress on confirm to watch the SSE.
    # -------------------------------------------------------------------

    @router.post("/api/rfcs/{slug}/graduate")
    async def graduate(slug: str, body: GraduateBody, request: Request) -> dict[str, Any]:
        viewer = auth.require_contributor(request)
        rfc = _require_super_draft(slug, viewer)
        # §13: only owners/arbiters of the RFC and app admins/owners may
        # graduate. Until §13.1's claim runs the entry has no owners, so
        # the set collapses to app admins/owners for unclaimed entries.
        if not _can_graduate(rfc, viewer):
            raise HTTPException(403, "Only RFC owners/arbiters or app admins/owners may graduate")

        # Refuse if an in-flight graduation is still running for this slug.
        existing = _get_active(slug)
        if existing is not None and not existing.finished:
            raise HTTPException(409, "Graduation already in progress for this slug")

        # §13.2 atomic re-validation. The dialog's debounced check runs
        # client-side as the admin types; this is the authoritative check
        # that closes the dialog-open-to-confirm race on the integer ID.
        # The id is OPTIONAL: a blank/absent id graduates with no number
        # (id stays null, slug is canonical per §2.3). Only a supplied id
        # is held to the regex and the collision re-check.
        rfc_id = (body.rfc_id or "").strip() or None
        owners = [o.strip() for o in body.owners if o.strip()]
        if not owners:
            raise HTTPException(422, "Add at least one initial owner")
        if rfc_id is not None:
            if not _is_valid_rfc_id(rfc_id):
                raise HTTPException(422, "ID must look like RFC-NNNN (at least four digits)")
            if _rfc_id_taken(rfc_id, excluding_slug=slug):
                raise HTTPException(409, f"Integer ID {rfc_id} is already taken")

        # Dual-read the meta-repo entry once (§22.4a sidecar-aware) — we need its
        # git state for the graduation commit and the body to carry through
        # unchanged (meta-only keeps the body in the entry, §13.3).
        st = await metadata_mod.read_entry_from_git(
            gitea, config.gitea_org,
            (projects_mod.default_content_repo(config) or ""), f"rfcs/{slug}.md",
        )
        if st is None:
            raise HTTPException(409, f"Meta entry rfcs/{slug}.md not found on main")
        super_draft_entry = st.entry

        arbiters = json.loads(rfc["arbiters_json"] or "[]") or owners[:1]

        # Compose the graduated frontmatter — body KEPT, graduation fields
        # filled, repo left null (§1). Serialized now so the PR-open step
        # has the contents pre-rendered.
        graduated_entry = entry_mod.Entry(
            slug=slug,
            title=super_draft_entry.title,
            state="active",
            id=rfc_id,
            repo=None,
            proposed_by=super_draft_entry.proposed_by,
            proposed_at=super_draft_entry.proposed_at,
            graduated_at=entry_mod.today(),
            graduated_by=viewer.gitea_login,
            owners=owners,
            arbiters=arbiters,
            tags=list(super_draft_entry.tags),
            models=super_draft_entry.models,
            funder=super_draft_entry.funder,
            body=super_draft_entry.body,
            # INV-7 (§22.4a): carry forward-compat / unknown frontmatter keys
            # through graduation rather than dropping them on the rebuild.
            extra=dict(super_draft_entry.extra),
        )
        graduation_files = metadata_mod.write_entry_files(
            f"rfcs/{slug}.md", graduated_entry, st)

        state = _new_active(
            slug, rfc_id=rfc_id, owners=owners, arbiters=arbiters,
        )

        # Audit: graduation started. The terminal `graduate_complete` /
        # `graduate_failed` rows below close the linkable sequence.
        _audit(
            viewer.user_id, viewer.gitea_login, "graduate_start",
            rfc_slug=slug,
            details={"rfc_id": rfc_id, "owners": owners},
        )

        # Test seam: `?_sync=1` awaits the orchestrator inline so
        # integration tests can assert post-conditions without driving
        # the SSE. Production clients POST then subscribe to the SSE.
        coro = _orchestrate(
            config=config, gitea=gitea, bot=bot,
            actor=viewer.as_actor(), state=state,
            graduation_files=graduation_files,
        )
        if request.query_params.get("_sync") == "1":
            await coro
        else:
            asyncio.create_task(coro)

        return {
            "ok": True,
            "slug": slug,
            "rfc_id": rfc_id,
            "stream_url": f"/api/rfcs/{slug}/graduate/progress",
            "finished": state.finished,
            "succeeded": state.succeeded,
        }

    # -------------------------------------------------------------------
    # GET /api/rfcs/<slug>/graduate/progress
    # SSE stream of the flip's step transitions (open_pr, merge_pr).
    # -------------------------------------------------------------------

    @router.get("/api/rfcs/{slug}/graduate/progress")
    async def graduate_progress(slug: str, request: Request):
        # The progress SSE surfaces admin-internal step detail (PR number)
        # that isn't part of the anonymous-read contract. POST /graduate is
        # gated to RFC owners/arbiters and app admins/owners; the read SSE
        # shares that operator-visible surface and requires an authenticated
        # viewer. We keep the floor at require_user (not require_contributor)
        # so a write-muted operator can still observe a graduation they
        # kicked off before being muted.
        auth.require_user(request)
        state = _get_active(slug)
        if state is None:
            raise HTTPException(404, "No graduation in flight for this slug")

        async def event_stream():
            yield _sse_event("snapshot", state.to_payload())
            if state.finished:
                yield _sse_event("done", state.to_payload())
                return
            while True:
                evt = await state.queue.get()
                if evt is None:
                    yield _sse_event("done", state.to_payload())
                    return
                yield _sse_event(evt.get("event", "update"), evt.get("payload"))

        headers = {"Cache-Control": "no-cache", "X-Accel-Buffering": "no"}
        return StreamingResponse(event_stream(), media_type="text/event-stream", headers=headers)

    # -------------------------------------------------------------------
    # §13.1: POST /api/rfcs/<slug>/claim
    # Opens a meta-repo PR adding the actor's gitea_login to the entry's
    # owners list. Anyone signed in may claim — the merge is gated to
    # owners/admins per §13.1.
    # -------------------------------------------------------------------

    @router.post("/api/rfcs/{slug}/claim")
    async def claim_ownership(slug: str, request: Request) -> dict[str, Any]:
        viewer = auth.require_contributor(request)
        rfc = _require_super_draft(slug, viewer)
        # §22.6/§22.7: claiming an unclaimed super-draft is a contribute action.
        # On a public project this is the pre-M2 baseline (an unclaimed entry
        # has no owners, so any granted contributor qualifies); on a gated
        # project it requires a project_contributor/admin grant.
        if not auth.can_contribute_to_rfc(viewer, slug):
            raise HTTPException(403, "You do not have contribute access to this project")
        existing_owners = json.loads(rfc["owners_json"] or "[]")
        if viewer.gitea_login in existing_owners:
            return {"ok": True, "noop": True}
        already = db.conn().execute(
            """
            SELECT pr_number FROM cached_prs
             WHERE rfc_slug = ? AND pr_kind = 'meta_claim' AND state = 'open'
            """,
            (slug,),
        ).fetchone()
        if already:
            raise HTTPException(409, f"A claim PR is already open: #{already['pr_number']}")

        st = await metadata_mod.read_entry_from_git(
            gitea, config.gitea_org,
            (projects_mod.default_content_repo(config) or ""), f"rfcs/{slug}.md",
        )
        if st is None:
            raise HTTPException(409, f"Meta entry rfcs/{slug}.md not found on main")
        if viewer.gitea_login in st.entry.owners:
            return {"ok": True, "noop": True}
        ent = metadata_mod.apply_values(
            st.entry, {"owners": st.entry.owners + [viewer.gitea_login]})
        files = metadata_mod.write_entry_files(f"rfcs/{slug}.md", ent, st)
        try:
            pr = await bot.open_claim_pr(
                viewer.as_actor(),
                org=config.gitea_org, meta_repo=(projects_mod.default_content_repo(config) or ""),
                slug=slug,
                files=files,
            )
        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"]}

    # -------------------------------------------------------------------
    # §13.7: POST /api/rfcs/<slug>/retire
    # Soft-delete an entry — flip its frontmatter `state` to `retired` via
    # an auto-merged meta-repo PR (§13.3 machinery reused). RFC owners and
    # site owners only — NOT app admins (§3.1). Allowed from super-draft or
    # active; the body and every other field (including the integer id) are
    # kept, so the entry stays recoverable in git and un-retire restores it
    # exactly.
    # -------------------------------------------------------------------

    @router.post("/api/rfcs/{slug}/retire")
    async def retire_rfc(slug: str, request: Request) -> dict[str, Any]:
        viewer = auth.require_contributor(request)
        rfc = _require_retirable(slug)
        if not _can_retire(rfc, viewer):
            raise HTTPException(
                403, "Only this RFC's owners or a site owner may retire it"
            )
        prior_state = rfc["state"]
        st = await _read_meta_entry(slug)
        entry = metadata_mod.apply_values(st.entry, {"state": "retired"})
        files = metadata_mod.write_entry_files(f"rfcs/{slug}.md", entry, st)
        await _run_state_flip(
            config=config, gitea=gitea, bot=bot, actor=viewer.as_actor(),
            slug=slug, files=files,
            verb="retire", target_state="retired",
        )
        _audit(
            viewer.user_id, viewer.gitea_login, "retire",
            rfc_slug=slug,
            details={"prior_state": prior_state, "rfc_id": entry.id},
        )
        await _refresh_catalog()
        return {"ok": True, "slug": slug, "state": "retired"}

    # -------------------------------------------------------------------
    # §13.7: POST /api/rfcs/<slug>/unretire
    # Bring a retired entry back to the state it held before retirement.
    # Site owners only (§3.1) — tighter than retire itself, so a soft-delete
    # is always recoverable by the operator but an RFC owner cannot reverse
    # their own retirement.
    # -------------------------------------------------------------------

    @router.post("/api/rfcs/{slug}/unretire")
    async def unretire_rfc(slug: str, request: Request) -> dict[str, Any]:
        viewer = auth.require_user(request)
        if viewer.role != "owner":
            raise HTTPException(403, "Only a site owner may un-retire an RFC")
        _require_retired(slug)
        restored = _prior_state_before_retire(slug)
        st = await _read_meta_entry(slug)
        entry = metadata_mod.apply_values(st.entry, {"state": restored})
        files = metadata_mod.write_entry_files(f"rfcs/{slug}.md", entry, st)
        await _run_state_flip(
            config=config, gitea=gitea, bot=bot, actor=viewer.as_actor(),
            slug=slug, files=files,
            verb="unretire", target_state=restored,
        )
        _audit(
            viewer.user_id, viewer.gitea_login, "unretire",
            rfc_slug=slug,
            details={"restored_state": restored},
        )
        await _refresh_catalog()
        return {"ok": True, "slug": slug, "state": restored}

    # -------------------------------------------------------------------
    # Helpers
    # -------------------------------------------------------------------

    def _require_super_draft(slug: str, viewer):
        row = db.conn().execute("SELECT *, (SELECT c.project_id FROM collections c WHERE c.id = cached_rfcs.collection_id) AS project_id FROM cached_rfcs WHERE slug = ?", (slug,)).fetchone()
        if row is None:
            raise HTTPException(404, "RFC not found")
        # §22.5 visibility gate (subtractive, §22.7): gated → 404 to non-members.
        auth.require_project_readable(viewer, row["project_id"])
        if row["state"] != "super-draft":
            raise HTTPException(409, f"RFC is {row['state']}, not super-draft")
        return row

    def _require_retirable(slug: str):
        row = db.conn().execute("SELECT *, (SELECT c.project_id FROM collections c WHERE c.id = cached_rfcs.collection_id) AS project_id FROM cached_rfcs WHERE slug = ?", (slug,)).fetchone()
        if row is None:
            raise HTTPException(404, "RFC not found")
        if row["state"] not in ("super-draft", "active"):
            raise HTTPException(409, f"RFC is {row['state']}, cannot be retired")
        return row

    def _require_retired(slug: str):
        row = db.conn().execute("SELECT *, (SELECT c.project_id FROM collections c WHERE c.id = cached_rfcs.collection_id) AS project_id FROM cached_rfcs WHERE slug = ?", (slug,)).fetchone()
        if row is None:
            raise HTTPException(404, "RFC not found")
        if row["state"] != "retired":
            raise HTTPException(409, f"RFC is {row['state']}, not retired")
        return row

    async def _read_meta_entry(slug: str):
        """Dual-read an entry from meta-main → EntryGitState (sidecar-aware,
        §22.4a). A migrated body-only `.md` reads cleanly; never raises on bad
        metadata (INV-3)."""
        st = await metadata_mod.read_entry_from_git(
            gitea, config.gitea_org,
            (projects_mod.default_content_repo(config) or ""), f"rfcs/{slug}.md",
        )
        if st is None:
            raise HTTPException(409, f"Meta entry rfcs/{slug}.md not found on main")
        return st

    async def _refresh_catalog() -> None:
        # Inline refresh so the catalog reflects the flip immediately; the
        # webhook/reconciler path is the steady-state per §4.1. A refresh
        # failure does not unwind the merge — the reconciler catches up.
        try:
            await cache.refresh_meta_repo(config, gitea)
            await cache.refresh_meta_branches(config, gitea)
            await cache.refresh_meta_pulls(config, gitea)
        except Exception as e:
            log.warning("retire/unretire cache refresh failed for slug: %s", e)

    return router


# ---------------------------------------------------------------------------
# Orchestrator — the §13.3 in-place flip
# ---------------------------------------------------------------------------


async def _orchestrate(
    *,
    config: Config,
    gitea: Gitea,
    bot: Bot,
    actor: Actor,
    state: GraduationState,
    graduation_files: list[dict],
) -> None:
    """Open the flip PR, then merge it. Two steps, no transaction:

      - open_pr fails  → nothing was created; the entry stays a super-draft.
      - merge_pr fails → close the open PR and delete its branch (the only
        artifact a mid-flip failure can leave on the meta repo), then the
        entry stays a super-draft.

    There is no rollback of a *merged* flip — once the meta-repo merge has
    landed, the path forward is §3's `withdraw` (§13.5).
    """
    try:
        # ----- Step 1: open the graduation PR (flip frontmatter) -----
        await _start(state, "open_pr", "Opening graduation PR…")
        try:
            pr = await bot.open_graduation_pr(
                actor,
                org=config.gitea_org, meta_repo=(projects_mod.default_content_repo(config) or ""),
                slug=state.slug,
                files=graduation_files,
                rfc_id=state.rfc_id,
                owners=state.owners,
            )
        except GiteaError as e:
            await _fail(state, "open_pr", f"Gitea: {e.detail}")
            await _finish_failed(state, failed_at="open_pr", on_behalf_of=actor.gitea_login)
            return
        state.new_pr_number = pr["number"]
        state.graduation_branch = pr["head"]["ref"]
        await _done(state, "open_pr", f"PR #{state.new_pr_number}")

        # ----- Step 2: merge the graduation PR -----
        await _start(state, "merge_pr", f"Merging PR #{state.new_pr_number}…")
        try:
            await bot.merge_graduation_pr(
                actor,
                org=config.gitea_org, meta_repo=(projects_mod.default_content_repo(config) or ""),
                pr_number=state.new_pr_number,
                head_branch=state.graduation_branch or "",
                slug=state.slug, rfc_id=state.rfc_id,
            )
        except GiteaError as e:
            await _fail(state, "merge_pr", f"Gitea: {e.detail}")
            await _cleanup_unmerged(config=config, bot=bot, actor=actor, state=state)
            await _finish_failed(state, failed_at="merge_pr", on_behalf_of=actor.gitea_login)
            return
        await _done(state, "merge_pr", f"PR #{state.new_pr_number} merged")

        # Refresh the cache so the catalog flips immediately. The webhook
        # flow is the steady-state path (§13.3); we refresh inline so the
        # dialog can transition to "graduation complete" with the catalog
        # row already showing `active`. A refresh failure does not unwind
        # the merge — the reconciler catches up per §4.1.
        try:
            await cache.refresh_meta_repo(config, gitea)
            await cache.refresh_meta_branches(config, gitea)
            await cache.refresh_meta_pulls(config, gitea)
        except Exception as e:
            log.warning("graduate cache refresh failed for %s: %s", state.slug, e)

        _audit(
            None, actor.gitea_login, "graduate_complete",
            rfc_slug=state.slug,
            details={
                "rfc_id": state.rfc_id,
                "owners": state.owners, "pr_number": state.new_pr_number,
            },
        )
        state.succeeded = True
        state.finished = True
        await state.queue.put({"event": "completed", "payload": state.to_payload()})
    except Exception as e:
        log.exception("graduate: unexpected error for %s", state.slug)
        running = next((s for s in state.steps if s.status == "running"), None)
        if running is not None:
            await _fail(state, running.key, f"unexpected: {e}")
        await _finish_failed(
            state, failed_at=running.key if running else "unknown",
            on_behalf_of=actor.gitea_login,
        )
    finally:
        # Push the sentinel so any open SSE handler returns.
        await state.queue.put(None)


async def _cleanup_unmerged(
    *, config: Config, bot: Bot, actor: Actor, state: GraduationState,
) -> None:
    """A merge failure leaves the flip PR open on its `graduate-<slug>-<hex>`
    branch. Close the PR and delete the branch so failed attempts don't
    accumulate on the meta repo. Best-effort — failures here are logged,
    not surfaced as a separate step (the entry already stays a super-draft).
    """
    if state.new_pr_number is None:
        return
    try:
        await bot.close_graduation_pr(
            actor,
            org=config.gitea_org, meta_repo=(projects_mod.default_content_repo(config) or ""),
            pr_number=state.new_pr_number,
            head_branch=state.graduation_branch or "",
            slug=state.slug, reason="graduation merge failed",
        )
    except Exception:
        log.exception("graduate cleanup: close PR #%s failed", state.new_pr_number)
    branch_name = state.graduation_branch or ""
    if branch_name:
        try:
            await bot.delete_branch(
                actor,
                owner=config.gitea_org,
                repo=(projects_mod.default_content_repo(config) or ""),
                branch=branch_name,
                slug=state.slug,
                action_kind="delete_post_merge_branch",
                reason="graduation merge failed",
            )
        except Exception:
            log.exception("graduate cleanup: delete_branch %s failed", branch_name)


async def _finish_failed(state: GraduationState, *, failed_at: str, on_behalf_of: str) -> None:
    """Mark any step after the failure as not-reached, write the audit
    row, and emit the terminal failed event."""
    seen_failure = False
    for s in state.steps:
        if s.status == "failed":
            seen_failure = True
            continue
        if seen_failure and s.status == "pending":
            s.status = "not-reached"
    _audit(
        None, on_behalf_of, "graduate_failed",
        rfc_slug=state.slug,
        details={
            "failed_at": failed_at,
            "error": state.error,
            "rfc_id": state.rfc_id,
            "pr_number": state.new_pr_number,
        },
    )
    state.finished = True
    state.succeeded = False
    await state.queue.put({"event": "failed", "payload": state.to_payload()})


# ---------------------------------------------------------------------------
# Permission + audit helpers
# ---------------------------------------------------------------------------


def _can_graduate(rfc, viewer) -> bool:
    if viewer is None:
        return False
    # §6.1 admin/owner or §22.6 project_admin OR §6.3 RFC owners/arbiters.
    if auth.is_collection_superuser(viewer, rfc["collection_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 _can_retire(rfc, viewer) -> bool:
    """§3.1: retire is narrower than graduate/withdraw — RFC owners
    (frontmatter) and site `owner`-role holders only, NOT app admins."""
    if viewer is None:
        return False
    if viewer.role == "owner":  # site owner
        return True
    owners = json.loads(rfc["owners_json"] or "[]")
    return viewer.gitea_login in owners


def _prior_state_before_retire(slug: str) -> str:
    """§13.7: the state to restore on un-retire. Read it from the most
    recent `retire` audit row's `prior_state`; if that's missing (e.g. the
    entry was retired out of band), fall back to inferring from the cached
    row — an assigned integer id implies it was `active`, else `super-draft`.
    """
    row = db.conn().execute(
        """
        SELECT details FROM actions
         WHERE rfc_slug = ? AND action_kind = 'retire'
         ORDER BY id DESC LIMIT 1
        """,
        (slug,),
    ).fetchone()
    if row and row["details"]:
        try:
            prior = json.loads(row["details"]).get("prior_state")
            if prior in ("super-draft", "active"):
                return prior
        except (ValueError, TypeError):
            pass
    cached = db.conn().execute(
        "SELECT rfc_id FROM cached_rfcs WHERE slug = ?", (slug,)
    ).fetchone()
    return "active" if (cached and cached["rfc_id"]) else "super-draft"


async def _run_state_flip(
    *,
    config: Config,
    gitea: Gitea,
    bot: Bot,
    actor: Actor,
    slug: str,
    files: list[dict],
    verb: str,
    target_state: str,
) -> None:
    """§13.7: open + merge a retire / un-retire state-flip PR. The flip is
    written to the entry's metadata sidecar (§22.4a) via `files` ops. Runs
    inline (no SSE — the flip is a single quick state change, unlike the
    multi-step graduation that streams progress). On an open failure
    nothing was created; on a merge failure the half-open PR/branch is
    cleaned up. Either failure raises 502 so the caller does not record the
    transition as having happened."""
    try:
        pr = await bot.open_retire_flip_pr(
            actor,
            org=config.gitea_org, meta_repo=(projects_mod.default_content_repo(config) or ""),
            slug=slug, files=files,
            verb=verb, target_state=target_state,
        )
    except GiteaError as e:
        raise HTTPException(502, f"Gitea: {e.detail}")
    pr_number = pr["number"]
    head_branch = pr["head"]["ref"]
    try:
        await bot.merge_retire_flip_pr(
            actor,
            org=config.gitea_org, meta_repo=(projects_mod.default_content_repo(config) or ""),
            pr_number=pr_number, head_branch=head_branch,
            slug=slug, verb=verb,
        )
    except GiteaError as e:
        # Clean up the unmerged flip PR/branch so failed attempts don't
        # accumulate (mirrors graduation's `_cleanup_unmerged`).
        try:
            await bot.close_graduation_pr(
                actor, org=config.gitea_org, meta_repo=(projects_mod.default_content_repo(config) or ""),
                pr_number=pr_number, head_branch=head_branch,
                slug=slug, reason=f"{verb} merge failed",
            )
            await bot.delete_branch(
                actor, owner=config.gitea_org, repo=(projects_mod.default_content_repo(config) or ""),
                branch=head_branch, slug=slug,
                action_kind="delete_post_merge_branch",
                reason=f"{verb} merge failed",
            )
        except Exception:
            log.exception("%s cleanup failed for %s", verb, slug)
        raise HTTPException(502, f"Gitea: {e.detail}")


def _audit(
    actor_user_id: int | None,
    on_behalf_of: str | None,
    action_kind: str,
    *,
    rfc_slug: str | None = None,
    branch_name: str | None = None,
    pr_number: int | None = None,
    details: dict | None = None,
) -> None:
    """Direct audit-log write for graduation lifecycle events that don't
    correspond to a single Gitea write. The per-step Gitea writes log
    themselves via the bot's `_log`; this is for the bracketing
    `graduate_start` / `graduate_complete` / `graduate_failed` rows."""
    db.conn().execute(
        """
        INSERT INTO actions
            (actor_user_id, on_behalf_of, action_kind, rfc_slug, branch_name, pr_number, bot_commit_sha, details)
        VALUES (?, ?, ?, ?, ?, ?, NULL, ?)
        """,
        (
            actor_user_id,
            on_behalf_of,
            action_kind,
            rfc_slug,
            branch_name,
            pr_number,
            json.dumps(details) if details else None,
        ),
    )
    # §15 chokepoint: the bracket rows drive their own notifications.
    from . import notify
    notify.fan_out_from_action(
        actor_user_id=actor_user_id,
        action_kind=action_kind,
        rfc_slug=rfc_slug,
        branch_name=branch_name,
        pr_number=pr_number,
        details=details,
    )


# ---------------------------------------------------------------------------
# Step state transitions
# ---------------------------------------------------------------------------


async def _start(state: GraduationState, key: str, detail: str) -> None:
    step = next(s for s in state.steps if s.key == key)
    step.status = "running"
    step.detail = detail
    await state.queue.put({"event": "step", "payload": state.to_payload()})


async def _done(state: GraduationState, key: str, detail: str) -> None:
    step = next(s for s in state.steps if s.key == key)
    step.status = "done"
    step.detail = detail
    await state.queue.put({"event": "step", "payload": state.to_payload()})


async def _fail(state: GraduationState, key: str, detail: str) -> None:
    step = next(s for s in state.steps if s.key == key)
    step.status = "failed"
    step.detail = detail
    state.error = detail
    await state.queue.put({"event": "step", "payload": state.to_payload()})


def _sse_event(name: str, payload: Any) -> str:
    return f"event: {name}\ndata: {json.dumps(payload)}\n\n"