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).

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
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
    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,
                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):
    rfc_id: str = Field(min_length=5, 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_project_superuser(viewer, rfc["project_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
        id_payload: dict[str, Any] = {"value": candidate_id, "ok": True, "error": None}
        if not candidate_id:
            id_payload["ok"] = False
            id_payload["error"] = "Integer ID is required"
        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)"
        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.
        rfc_id = body.rfc_id.strip()
        owners = [o.strip() for o in body.owners if o.strip()]
        if not owners:
            raise HTTPException(422, "Add at least one initial owner")
        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")

        # Read the meta-repo entry once — we need the file's sha for the
        # graduation PR's update_file call and the body to carry through
        # unchanged (meta-only keeps the body in the entry, §13.3).
        fetched = await gitea.read_file(
            config.gitea_org, config.meta_repo, f"rfcs/{slug}.md", ref="main",
        )
        if fetched is None:
            raise HTTPException(409, f"Meta entry rfcs/{slug}.md not found on main")
        meta_text, meta_sha = fetched
        try:
            super_draft_entry = entry_mod.parse(meta_text)
        except Exception as e:
            raise HTTPException(500, f"Meta entry malformed: {e}")

        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,
        )
        graduated_contents = entry_mod.serialize(graduated_entry)

        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,
            graduated_contents=graduated_contents,
            meta_file_sha=meta_sha,
        )
        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']}")

        fetched = await gitea.read_file(
            config.gitea_org, config.meta_repo, f"rfcs/{slug}.md", ref="main",
        )
        if fetched is None:
            raise HTTPException(409, f"Meta entry rfcs/{slug}.md not found on main")
        meta_text, meta_sha = fetched
        try:
            ent = entry_mod.parse(meta_text)
        except Exception as e:
            raise HTTPException(500, f"Meta entry malformed: {e}")
        if viewer.gitea_login in ent.owners:
            return {"ok": True, "noop": True}
        ent.owners = ent.owners + [viewer.gitea_login]
        new_contents = entry_mod.serialize(ent)
        try:
            pr = await bot.open_claim_pr(
                viewer.as_actor(),
                org=config.gitea_org, meta_repo=config.meta_repo,
                slug=slug,
                new_file_contents=new_contents, prior_sha=meta_sha,
            )
        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"]}

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

    def _require_super_draft(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): 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

    return router


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


async def _orchestrate(
    *,
    config: Config,
    gitea: Gitea,
    bot: Bot,
    actor: Actor,
    state: GraduationState,
    graduated_contents: str,
    meta_file_sha: str,
) -> 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=config.meta_repo,
                slug=state.slug,
                new_file_contents=graduated_contents,
                prior_sha=meta_file_sha,
                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=config.meta_repo,
                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=config.meta_repo,
            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=config.meta_repo,
                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_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 _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"