rfc-app/backend/app/bot.py

"""The bot wrapper.

Per §1: the bot service account is the only Git writer in the system.
Per §6.5: every commit, branch creation, and PR merge carries an
On-behalf-of: trailer naming the acting user.

This module is the single chokepoint. Every write to Gitea — file
creation, branch creation, PR open, PR merge, PR close — flows through
a Bot method that takes an `actor` (the authenticated user whose gesture
produced the action) and an `action_kind` (one of the values recorded
in the `actions` table). The wrapper:

  - calls the Gitea HTTP client with the bot's credentials,
  - appends the trailer to commit/PR/comment bodies,
  - records a row in `actions` so the app's accountability surface and
    the Git log carry the same record.

If you find yourself wanting to import gitea.py directly to perform a
write, the spec is right and you are wrong: the wrapper is the
invariant. Read operations live in `gitea.py` and can be called from
anywhere.
"""
from __future__ import annotations

import asyncio
import json
import logging
from dataclasses import dataclass

from . import db, entry as entry_mod, notify
from .gitea import Gitea, GiteaError

log = logging.getLogger(__name__)


_MERGE_TRANSIENT_HINTS = (
    "please try again later",
    "is not ready",
    "not ready to be merged",
)


async def _merge_with_retry(
    gitea: Gitea,
    org: str,
    meta_repo: str,
    pr_number: int,
    *,
    merge_message_title: str,
    merge_message_body: str,
    attempts: int = 3,
    delay_seconds: float = 1.0,
) -> None:
    """Call gitea.merge_pull and retry on Gitea's transient
    "Please try again later" / "not ready" responses. After
    `wait_for_mergeable` returns this is rare, but Gitea has been
    observed to flip back to the unready state momentarily on a
    write-amplified instance; a small bounded retry closes the gap.

    Any other GiteaError propagates immediately so the §13.3
    orchestrator can fail the step and run rollback.
    """
    last_error: GiteaError | None = None
    for attempt in range(1, attempts + 1):
        try:
            await gitea.merge_pull(
                org, meta_repo, pr_number,
                merge_message_title=merge_message_title,
                merge_message_body=merge_message_body,
                style="merge",
            )
            return
        except GiteaError as e:
            detail = (e.detail or "").lower()
            if not any(hint in detail for hint in _MERGE_TRANSIENT_HINTS):
                raise
            last_error = e
            log.warning(
                "merge_pull transient (attempt %d/%d) for %s/%s#%d: %s",
                attempt, attempts, org, meta_repo, pr_number, e.detail,
            )
            if attempt < attempts:
                await asyncio.sleep(delay_seconds)
    assert last_error is not None
    raise last_error


@dataclass(frozen=True)
class Actor:
    """The user whose gesture is producing a write."""
    user_id: int
    gitea_login: str
    display_name: str
    email: str


def _trailer(actor: Actor) -> str:
    return f"On-behalf-of: {actor.display_name} <{actor.gitea_login}>"


def _stamp(message_subject: str, message_body: str, actor: Actor) -> tuple[str, str]:
    """Compose subject + body with the On-behalf-of trailer appended.

    Subject and body are returned separately because Gitea's merge API
    takes them on distinct fields; for file commits we hand back a
    single string in the caller.
    """
    body = message_body.rstrip()
    trailer = _trailer(actor)
    if body:
        return message_subject, f"{body}\n\n{trailer}"
    return message_subject, trailer


def _stamp_single(message: str, actor: Actor) -> str:
    subject, _, rest = message.partition("\n")
    subject, body = _stamp(subject, rest.lstrip(), actor)
    return f"{subject}\n\n{body}".rstrip()


def _log(
    actor: Actor,
    action_kind: str,
    *,
    rfc_slug: str | None = None,
    branch_name: str | None = None,
    pr_number: int | None = None,
    bot_commit_sha: str | None = None,
    details: dict | None = None,
) -> None:
    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 (?, ?, ?, ?, ?, ?, ?, ?)
        """,
        (
            actor.user_id,
            actor.gitea_login,
            action_kind,
            rfc_slug,
            branch_name,
            pr_number,
            bot_commit_sha,
            json.dumps(details) if details else None,
        ),
    )
    # §15 chokepoint per §19.1 brief: fan-out runs inline after the
    # audit row lands. notify.py owns the routing rules and the
    # auto-watch upsert per §15.6; the call is intentionally a
    # single line here so the chokepoint is one place to read.
    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,
    )


class Bot:
    def __init__(self, gitea: Gitea):
        self._gitea = gitea

    # ----- Content repo: collection structure (§22 S2) -----

    async def create_collection(
        self,
        actor: Actor,
        *,
        org: str,
        content_repo: str,
        collection_id: str,
        manifest_yaml: str,
    ) -> dict:
        """§22 S2: commit `<collection_id>/.collection.yaml` to the content
        repo's main. A structural admin action — committed straight to main (no
        PR), like the registry config it feeds; the registry mirror then upserts
        the collections row (§22.2 keeps the registry the source of truth). Logs
        an audit row for the §6.5 trail."""
        path = f"{collection_id}/.collection.yaml"
        created = await self._gitea.create_file(
            org,
            content_repo,
            path,
            content=manifest_yaml,
            message=_stamp_single(f"chore: create collection {collection_id}", actor),
            branch="main",
            author_name=actor.display_name,
            author_email=actor.email or f"{actor.gitea_login}@users.noreply",
        )
        _log(
            actor,
            "create_collection",
            bot_commit_sha=created.get("commit", {}).get("sha"),
            details={"collection_id": collection_id, "repo": content_repo},
        )
        return created

    async def create_project(
        self,
        actor: Actor,
        *,
        org: str,
        registry_repo: str,
        content_repo: str,
        project_id: str,
        projects_yaml_new: str,
        projects_yaml_sha: str,
        readme_text: str,
    ) -> dict:
        """§22 S5 (§A.2): stand up a new project. A global-Owner action wrapping
        a bot write at two git sources:

          1. **provision the content repo** — create `org/content_repo` if it
             doesn't exist, then seed a `README.md` on `main` so the branch
             exists (the contents API initialises the repo with that commit; the
             corpus mirror and the propose path both need a `main` to write to).
          2. **register the project** — commit the caller-composed
             `projects.yaml` (the existing doc with the new project appended) to
             the registry repo's `main`.

        Like `create_collection`, this is a structural admin action committed
        straight to main (no PR), like the registry config it feeds; the caller
        then re-runs the registry mirror so the new `projects` + default
        `collections` rows flow from the registry (§22.2 keeps the registry the
        source of truth). Logs a `create_project` audit row for the §6.5 trail.
        Returns the registry update_file result (carries the new commit sha)."""
        ae = actor.email or f"{actor.gitea_login}@users.noreply"
        existing = await self._gitea.get_repo(org, content_repo)
        if existing is None:
            await self._gitea.create_org_repo(
                org, content_repo, description=f"Content repo for project {project_id}"
            )
        # Seed a README if absent, which also establishes `main` on a freshly
        # created (auto_init=False) repo — the contents API initialises the repo
        # with that commit. Keyed on the README rather than the branch so it is
        # idempotent and behaves identically whether the repo has a bare `main`
        # or no branch at all. Mirrors `ensure_rfc_repo_seed`'s empty-repo seed.
        readme = await self._gitea.get_contents(org, content_repo, "README.md", ref="main")
        if readme is None:
            await self._gitea.create_file(
                org,
                content_repo,
                "README.md",
                content=readme_text,
                message=_stamp_single(f"chore: initialise content repo for {project_id}", actor),
                branch="main",
                author_name=actor.display_name,
                author_email=ae,
            )
        result = await self._gitea.update_file(
            org,
            registry_repo,
            "projects.yaml",
            content=projects_yaml_new,
            sha=projects_yaml_sha,
            message=_stamp_single(f"chore: create project {project_id}", actor),
            branch="main",
            author_name=actor.display_name,
            author_email=ae,
        )
        commit_sha = (
            result.get("commit", {}).get("sha")
            or result.get("content", {}).get("sha")
            or ""
        )
        _log(
            actor,
            "create_project",
            bot_commit_sha=commit_sha,
            details={"project_id": project_id, "content_repo": content_repo},
        )
        return result

    # ----- Meta repo: idea PRs (§9.1 / §9.2) -----

    async def open_idea_pr(
        self,
        actor: Actor,
        *,
        org: str,
        meta_repo: str,
        slug: str,
        file_contents: str,
        pr_title: str,
        pr_description: str,
        rfcs_dir: str = "rfcs",
    ) -> dict:
        """Per §9.1: open a meta-repo PR adding one file under `<rfcs_dir>/`.

        One file per PR keeps idea submissions atomic and conflict-free.
        The PR title and the file-add commit subject share §9.2's fixed
        pattern; callers compose `pr_title` as `Propose: <Title>`. §22 S2:
        `rfcs_dir` carries the target collection's `<subfolder>/rfcs` so a
        propose into a named collection writes under its subfolder; it
        defaults to `rfcs` (the default collection / shipped behaviour).
        """
        branch = f"propose/{slug}"
        await self._gitea.create_branch(org, meta_repo, branch, from_branch="main")
        commit_subject = pr_title  # §9.2: shared pattern
        commit_message = _stamp_single(commit_subject, actor)
        created = await self._gitea.create_file(
            org,
            meta_repo,
            f"{rfcs_dir}/{slug}.md",
            content=file_contents,
            message=commit_message,
            branch=branch,
            author_name=actor.display_name,
            author_email=actor.email or f"{actor.gitea_login}@users.noreply",
        )
        commit_sha = created.get("commit", {}).get("sha")
        pr_body_subject, pr_body = _stamp("", pr_description, actor)
        del pr_body_subject  # only the body matters here
        pr = await self._gitea.create_pull(
            org,
            meta_repo,
            title=pr_title,
            body=pr_body,
            head=branch,
            base="main",
        )
        _log(
            actor,
            "propose_rfc",
            rfc_slug=slug,
            branch_name=branch,
            pr_number=pr["number"],
            bot_commit_sha=commit_sha,
            details={"pr_title": pr_title},
        )
        return pr

    async def merge_idea_pr(
        self,
        actor: Actor,
        *,
        org: str,
        meta_repo: str,
        pr_number: int,
        slug: str,
    ) -> None:
        """Per §9.3: owner/admin merges an idea PR, creating the super-draft."""
        subject = f"Merge proposal: {slug}"
        body = _trailer(actor)
        await self._gitea.merge_pull(
            org,
            meta_repo,
            pr_number,
            merge_message_title=subject,
            merge_message_body=body,
        )
        _log(
            actor,
            "merge_proposal",
            rfc_slug=slug,
            pr_number=pr_number,
        )

    async def decline_idea_pr(
        self,
        actor: Actor,
        *,
        org: str,
        meta_repo: str,
        pr_number: int,
        slug: str,
        comment: str,
    ) -> None:
        """Per §9.3: owner/admin declines an idea PR with a required comment.

        The comment is posted to the PR (the durable Git artifact) and a
        mirroring system-author thread_messages row is written by the
        caller so the chat record carries the act inline.
        """
        commented = comment.strip() or "(no comment provided)"
        body = f"{commented}\n\n{_trailer(actor)}"
        await self._gitea.create_issue_comment(org, meta_repo, pr_number, body)
        await self._gitea.close_pull(org, meta_repo, pr_number)
        _log(
            actor,
            "decline_proposal",
            rfc_slug=slug,
            pr_number=pr_number,
            details={"comment": commented},
        )

    async def withdraw_idea_pr(
        self,
        actor: Actor,
        *,
        org: str,
        meta_repo: str,
        pr_number: int,
        slug: str,
    ) -> None:
        await self._gitea.close_pull(org, meta_repo, pr_number)
        _log(
            actor,
            "withdraw_proposal",
            rfc_slug=slug,
            pr_number=pr_number,
        )

    # ----- Meta repo: metadata-pane PRs (§9.5) -----

    async def open_metadata_pr(
        self,
        actor: Actor,
        *,
        org: str,
        meta_repo: str,
        slug: str,
        new_file_contents: str,
        prior_sha: str,
        pr_title: str,
        pr_description: str,
    ) -> dict:
        """Per §9.5: a metadata-pane edit (title or tags) on a super-draft
        opens a tiny meta-repo PR that touches only the frontmatter of
        `rfcs/<slug>.md`. One commit, one PR, easy to triage. The branch
        name uses the dash-separated `metadata-<slug>-<6hex>` shape — same
        routing-friendly form Slice 4 picked for edit branches per the
        §19.2 path-routing candidate.
        """
        import secrets

        branch = f"metadata-{slug}-{secrets.token_hex(3)}"
        await self._gitea.create_branch(org, meta_repo, branch, from_branch="main")
        commit_subject = pr_title
        commit_message = _stamp_single(commit_subject, actor)
        result = await self._gitea.update_file(
            org,
            meta_repo,
            f"rfcs/{slug}.md",
            content=new_file_contents,
            sha=prior_sha,
            message=commit_message,
            branch=branch,
            author_name=actor.display_name,
            author_email=actor.email or f"{actor.gitea_login}@users.noreply",
        )
        commit_sha = result.get("commit", {}).get("sha") or result.get("content", {}).get("sha") or ""
        _subject, pr_body = _stamp("", pr_description, actor)
        pr = await self._gitea.create_pull(
            org,
            meta_repo,
            title=pr_title,
            body=pr_body,
            head=branch,
            base="main",
        )
        _log(
            actor,
            "open_metadata_pr",
            rfc_slug=slug,
            branch_name=branch,
            pr_number=pr["number"],
            bot_commit_sha=commit_sha,
            details={"pr_title": pr_title},
        )
        return pr

    # ----- Per-RFC repo: branches (§8.3, §8.14) -----

    async def cut_branch_from_main(
        self,
        actor: Actor,
        *,
        owner: str,
        repo: str,
        new_branch: str,
        slug: str,
        from_branch: str = "main",
    ) -> dict:
        """Per §8.14: 'Start Contributing' on main cuts a new branch.

        Also covers the §8.3 case of a contributor wanting a fresh branch
        for a piece of work. Returns the Gitea branch payload.
        """
        created = await self._gitea.create_branch(owner, repo, new_branch, from_branch=from_branch)
        _log(
            actor,
            "create_branch",
            rfc_slug=slug,
            branch_name=new_branch,
            details={"from": from_branch, "repo": f"{owner}/{repo}"},
        )
        return created

    # ----- Per-RFC repo: per-accepted-change commits (§8.6, §8.9) -----

    async def commit_accepted_change(
        self,
        actor: Actor,
        *,
        owner: str,
        repo: str,
        branch: str,
        file_path: str,
        new_content: str,
        prior_sha: str,
        change_id: int,
        original: str,
        proposed: str,
        ai_proposed: str | None,
        reason: str,
        source_message_id: int | None,
        slug: str,
    ) -> str:
        """Per §8.6: one commit per accepted change.

        The commit message subject is a short structural description; the
        body carries `original`, `proposed`, and `reason` in named
        sections. When the contributor edited the AI's proposal before
        accepting (§8.9's `was_edited_before_accept`), the AI's original
        wording is preserved under an `AI proposed:` section so the
        timeline records both what was offered and what landed.

        Trailers: `Change-Id`, `Source-Message-Id` (where applicable),
        and the standard `On-behalf-of:` per §6.5.

        Returns the commit SHA.
        """
        subject = _subject_from_reason(reason, fallback="Accept change")
        body_lines = [
            "**Original:**",
            original.strip(),
            "",
            "**Proposed:**",
            proposed.strip(),
        ]
        if ai_proposed is not None and ai_proposed.strip() != proposed.strip():
            body_lines += ["", "**AI proposed (edited before accept):**", ai_proposed.strip()]
        if reason and reason.strip():
            body_lines += ["", "**Reason:**", reason.strip()]
        body_lines += ["", f"Change-Id: {change_id}"]
        if source_message_id is not None:
            body_lines += [f"Source-Message-Id: {source_message_id}"]
        body_lines += [_trailer(actor)]
        message = subject + "\n\n" + "\n".join(body_lines).strip()

        result = await self._gitea.update_file(
            owner,
            repo,
            file_path,
            content=new_content,
            sha=prior_sha,
            message=message,
            branch=branch,
            author_name=actor.display_name,
            author_email=actor.email or f"{actor.gitea_login}@users.noreply",
        )
        sha = result.get("commit", {}).get("sha") or result.get("content", {}).get("sha") or ""
        _log(
            actor,
            "accept_change",
            rfc_slug=slug,
            branch_name=branch,
            bot_commit_sha=sha,
            details={"change_id": change_id, "file_path": file_path},
        )
        return sha

    # ----- Per-RFC repo: manual-edit flushes (§8.6, §8.11) -----

    async def commit_manual_flush(
        self,
        actor: Actor,
        *,
        owner: str,
        repo: str,
        branch: str,
        file_path: str,
        new_content: str,
        prior_sha: str,
        change_id: int,
        paragraph_count: int,
        slug: str,
    ) -> str:
        """Per §8.6 / §8.11: one commit per manual-edit flush window.

        Subject names the structural extent so a reviewer scanning the
        log can size the change at a glance; the body carries the
        change-id trailer that binds the commit to the resolved card in
        the panel.
        """
        plural = "" if paragraph_count == 1 else "s"
        subject = f"manual edit: {paragraph_count} paragraph{plural}"
        body_lines = [
            f"Change-Id: {change_id}",
            _trailer(actor),
        ]
        message = subject + "\n\n" + "\n".join(body_lines)
        result = await self._gitea.update_file(
            owner,
            repo,
            file_path,
            content=new_content,
            sha=prior_sha,
            message=message,
            branch=branch,
            author_name=actor.display_name,
            author_email=actor.email or f"{actor.gitea_login}@users.noreply",
        )
        sha = result.get("commit", {}).get("sha") or result.get("content", {}).get("sha") or ""
        _log(
            actor,
            "manual_flush",
            rfc_slug=slug,
            branch_name=branch,
            bot_commit_sha=sha,
            details={"change_id": change_id, "paragraph_count": paragraph_count},
        )
        return sha

    # ----- Per-RFC repo: seeding (test/dev fixtures, future graduation) -----

    # ----- Per-RFC repo: PRs (§10) -----

    async def open_branch_pr(
        self,
        actor: Actor,
        *,
        owner: str,
        repo: str,
        head_branch: str,
        title: str,
        description: str,
        slug: str,
        supersedes_pr_number: int | None = None,
    ) -> dict:
        """Per §10.1: open a PR from a branch against main.

        The PR's body carries the contributor's description, optional
        `Supersedes:` trailer for the §10.9 replay path, and the
        standard `On-behalf-of:` trailer per §6.5.
        """
        body_lines = [description.strip()]
        if supersedes_pr_number is not None:
            body_lines += ["", f"Supersedes: #{supersedes_pr_number}"]
        body_lines += ["", _trailer(actor)]
        body = "\n".join(body_lines).strip()
        pr = await self._gitea.create_pull(
            owner,
            repo,
            title=title,
            body=body,
            head=head_branch,
            base="main",
        )
        _log(
            actor,
            "open_branch_pr",
            rfc_slug=slug,
            branch_name=head_branch,
            pr_number=pr["number"],
            details={
                "title": title,
                "supersedes": supersedes_pr_number,
                "repo": f"{owner}/{repo}",
            },
        )
        return pr

    async def merge_branch_pr(
        self,
        actor: Actor,
        *,
        owner: str,
        repo: str,
        pr_number: int,
        head_branch: str,
        slug: str,
    ) -> None:
        """Per §10.5: no-fast-forward merge.

        Gitea's `style='merge'` produces a merge commit; the
        per-acceptance commits from §8.6 remain individually reachable
        in main's history. The merge commit's body records the merging
        user via the `On-behalf-of:` trailer — the merge commit's
        author stays the bot (the bot is the only Git writer per §1)
        but the trailer carries the human accountability.
        """
        subject = f"Merge branch '{head_branch}'"
        body = _trailer(actor)
        await self._gitea.merge_pull(
            owner,
            repo,
            pr_number,
            merge_message_title=subject,
            merge_message_body=body,
            style="merge",
        )
        _log(
            actor,
            "merge_branch_pr",
            rfc_slug=slug,
            branch_name=head_branch,
            pr_number=pr_number,
            details={"repo": f"{owner}/{repo}"},
        )

    async def withdraw_branch_pr(
        self,
        actor: Actor,
        *,
        owner: str,
        repo: str,
        pr_number: int,
        head_branch: str,
        slug: str,
        reason: str = "withdraw",
    ) -> None:
        """Per §10.8: close the PR; do not delete the branch."""
        await self._gitea.close_pull(owner, repo, pr_number)
        _log(
            actor,
            "withdraw_branch_pr" if reason == "withdraw" else "supersede_branch_pr",
            rfc_slug=slug,
            branch_name=head_branch,
            pr_number=pr_number,
            details={"repo": f"{owner}/{repo}", "reason": reason},
        )

    async def cut_resolution_branch(
        self,
        actor: Actor,
        *,
        owner: str,
        repo: str,
        original_branch: str,
        resolution_branch: str,
        slug: str,
    ) -> dict:
        """Per §10.9: cut a fresh branch off main's tip into which the
        original branch's changes will be replayed. The bot owns the
        cut; the replay itself is a sequence of commit_accepted_change
        / manual flush operations driven by the API layer."""
        created = await self._gitea.create_branch(
            owner, repo, resolution_branch, from_branch="main"
        )
        _log(
            actor,
            "create_resolution_branch",
            rfc_slug=slug,
            branch_name=resolution_branch,
            details={
                "repo": f"{owner}/{repo}",
                "original_branch": original_branch,
            },
        )
        return created

    async def commit_replay_change(
        self,
        actor: Actor,
        *,
        owner: str,
        repo: str,
        branch: str,
        file_path: str,
        new_content: str,
        prior_sha: str,
        original_change_id: int,
        original: str,
        proposed: str,
        reason: str,
        slug: str,
    ) -> str:
        """Per §10.9: a single replayed accept lands as its own commit on
        the resolution branch, so the §8.6 evidence shape is preserved.
        The subject mirrors `commit_accepted_change`'s but the body
        records the original change id so the resolution PR's
        conversation can stitch back to the original branch's chat."""
        subject = _subject_from_reason(reason, fallback="Replay change")
        body_lines = [
            "**Original:**",
            original.strip(),
            "",
            "**Proposed:**",
            proposed.strip(),
        ]
        if reason and reason.strip():
            body_lines += ["", "**Reason:**", reason.strip()]
        body_lines += ["", f"Replayed-Change-Id: {original_change_id}"]
        body_lines += [_trailer(actor)]
        message = subject + "\n\n" + "\n".join(body_lines).strip()
        result = await self._gitea.update_file(
            owner,
            repo,
            file_path,
            content=new_content,
            sha=prior_sha,
            message=message,
            branch=branch,
            author_name=actor.display_name,
            author_email=actor.email or f"{actor.gitea_login}@users.noreply",
        )
        sha = result.get("commit", {}).get("sha") or result.get("content", {}).get("sha") or ""
        _log(
            actor,
            "replay_change",
            rfc_slug=slug,
            branch_name=branch,
            bot_commit_sha=sha,
            details={"original_change_id": original_change_id, "file_path": file_path},
        )
        return sha

    # ----- §13 graduation (meta-only): open + merge the flip PR -----

    async def open_graduation_pr(
        self,
        actor: Actor,
        *,
        org: str,
        meta_repo: str,
        slug: str,
        new_file_contents: str,
        prior_sha: str,
        rfc_id: str | None,
        owners: list[str],
    ) -> dict:
        """§13.3 (meta-only): open a PR against the meta repo that flips the
        entry's frontmatter to `state: active` with the graduation stamps
        and — **optionally** — the integer `id`, **keeping the body
        unchanged** (§1 meta-only topology; no repo is created and no body
        is stripped). When `rfc_id` is None the entry graduates without a
        number (id stays null, slug is canonical per §2.3, §13.2). Branch
        name uses the `graduate-<slug>-<6hex>` shape — dash-separated like
        the other meta-repo branches per the §19.2 path-routing candidate.
        """
        import secrets

        branch = f"graduate-{slug}-{secrets.token_hex(3)}"
        await self._gitea.create_branch(org, meta_repo, branch, from_branch="main")
        ae = actor.email or f"{actor.gitea_login}@users.noreply"
        commit_subject = f"Graduate {slug} → {rfc_id}" if rfc_id else f"Graduate {slug} (no number)"
        commit_message = _stamp_single(commit_subject, actor)
        result = await self._gitea.update_file(
            org, meta_repo, f"rfcs/{slug}.md",
            content=new_file_contents,
            sha=prior_sha,
            message=commit_message,
            branch=branch,
            author_name=actor.display_name, author_email=ae,
        )
        commit_sha = (
            result.get("commit", {}).get("sha")
            or result.get("content", {}).get("sha")
            or ""
        )
        pr_title = f"Graduate {slug} → {rfc_id}" if rfc_id else f"Graduate {slug} (no number)"
        owners_str = ", ".join(owners) if owners else "(none)"
        id_line = f"- ID: `{rfc_id}`\n" if rfc_id else "- ID: (none — identified by slug)\n"
        pr_body_text = (
            f"Graduates super-draft `{slug}` to active.\n\n"
            f"{id_line}"
            f"- Owners: {owners_str}\n\n"
            f"This is an in-place state flip per the meta-only topology\n"
            f"(SPEC §1, §13.3): the entry `rfcs/{slug}.md` keeps its body and\n"
            f"stays in the meta repo. Only the frontmatter changes — `state`,\n"
            f"`id`, and the graduation stamps."
        )
        _subject, pr_body = _stamp("", pr_body_text, actor)
        pr = await self._gitea.create_pull(
            org, meta_repo,
            title=pr_title, body=pr_body, head=branch, base="main",
        )
        _log(
            actor,
            "graduate_pr_open",
            rfc_slug=slug,
            branch_name=branch,
            pr_number=pr["number"],
            bot_commit_sha=commit_sha,
            details={"pr_title": pr_title, "rfc_id": rfc_id},
        )
        return pr

    async def merge_graduation_pr(
        self,
        actor: Actor,
        *,
        org: str,
        meta_repo: str,
        pr_number: int,
        head_branch: str,
        slug: str,
        rfc_id: str | None,
    ) -> None:
        """§13.3 step 4: auto-merge the graduation PR with the admin as
        merge actor. Distinct action_kind so the audit log carries the
        graduation as a linkable sequence per §13.3's transactional shape.

        Gitea computes a PR's mergeability asynchronously in a background
        job. Step 3 (`open_graduation_pr`) returns the moment the PR is
        created, which is typically before that job has finished — calling
        `merge_pull` immediately produces a `405 "Please try again later"`
        and the §13.3 orchestrator interprets it as a terminal failure.
        We wait for the mergeability computation to settle before the
        merge call, then retry the merge a couple of times if Gitea still
        reports the transient state (belt-and-suspenders against the same
        race appearing in a different shape).
        """
        subject = f"Graduate {slug} → {rfc_id}" if rfc_id else f"Graduate {slug} (no number)"
        body = _trailer(actor)

        try:
            await self._gitea.wait_for_mergeable(org, meta_repo, pr_number)
        except TimeoutError as e:
            raise GiteaError(409, str(e)) from e

        await _merge_with_retry(
            self._gitea, org, meta_repo, pr_number,
            merge_message_title=subject, merge_message_body=body,
        )
        _log(
            actor,
            "graduate_pr_merge",
            rfc_slug=slug,
            branch_name=head_branch,
            pr_number=pr_number,
            details={"rfc_id": rfc_id},
        )

    # ----- §13.7 retire / un-retire: open + merge a state-flip PR -----

    async def open_retire_flip_pr(
        self,
        actor: Actor,
        *,
        org: str,
        meta_repo: str,
        slug: str,
        new_file_contents: str,
        prior_sha: str,
        verb: str,
        target_state: str,
    ) -> dict:
        """§13.7: open a PR flipping `rfcs/<slug>.md` to `state:
        <target_state>` — for retire (`verb='retire'`, target `retired`)
        or un-retire (`verb='unretire'`, target the restored prior state).
        Only the frontmatter `state` changes; the body and every other
        field (including the integer `id`) are kept, so an un-retire
        restores the entry exactly. Branch shape mirrors graduation's
        `<verb>-<slug>-<6hex>`.
        """
        import secrets

        branch = f"{verb}-{slug}-{secrets.token_hex(3)}"
        await self._gitea.create_branch(org, meta_repo, branch, from_branch="main")
        ae = actor.email or f"{actor.gitea_login}@users.noreply"
        verb_title = "Retire" if verb == "retire" else "Un-retire"
        commit_subject = f"{verb_title} {slug}"
        commit_message = _stamp_single(commit_subject, actor)
        result = await self._gitea.update_file(
            org, meta_repo, f"rfcs/{slug}.md",
            content=new_file_contents,
            sha=prior_sha,
            message=commit_message,
            branch=branch,
            author_name=actor.display_name, author_email=ae,
        )
        commit_sha = (
            result.get("commit", {}).get("sha")
            or result.get("content", {}).get("sha")
            or ""
        )
        pr_title = f"{verb_title} {slug}"
        pr_body_text = (
            f"{verb_title}s `{slug}` (state → `{target_state}`).\n\n"
            f"This is an in-place frontmatter flip per SPEC §13.7 — the\n"
            f"entry `rfcs/{slug}.md` keeps its body and every other field;\n"
            f"only `state` changes."
        )
        _subject, pr_body = _stamp("", pr_body_text, actor)
        pr = await self._gitea.create_pull(
            org, meta_repo,
            title=pr_title, body=pr_body, head=branch, base="main",
        )
        _log(
            actor,
            f"{verb}_pr_open",
            rfc_slug=slug,
            branch_name=branch,
            pr_number=pr["number"],
            bot_commit_sha=commit_sha,
            details={"pr_title": pr_title, "target_state": target_state},
        )
        return pr

    async def merge_retire_flip_pr(
        self,
        actor: Actor,
        *,
        org: str,
        meta_repo: str,
        pr_number: int,
        head_branch: str,
        slug: str,
        verb: str,
    ) -> None:
        """§13.7: auto-merge the retire / un-retire flip PR (same
        mergeable-wait + retry shape as `merge_graduation_pr`)."""
        verb_title = "Retire" if verb == "retire" else "Un-retire"
        subject = f"{verb_title} {slug}"
        body = _trailer(actor)
        try:
            await self._gitea.wait_for_mergeable(org, meta_repo, pr_number)
        except TimeoutError as e:
            raise GiteaError(409, str(e)) from e
        await _merge_with_retry(
            self._gitea, org, meta_repo, pr_number,
            merge_message_title=subject, merge_message_body=body,
        )
        _log(
            actor,
            f"{verb}_pr_merge",
            rfc_slug=slug,
            branch_name=head_branch,
            pr_number=pr_number,
            details={},
        )

    # ----- §13.3 (meta-only): cleanup of an unmerged flip PR -----

    async def close_graduation_pr(
        self,
        actor: Actor,
        *,
        org: str,
        meta_repo: str,
        pr_number: int,
        head_branch: str,
        slug: str,
        reason: str,
    ) -> None:
        """Undo of `open_graduation_pr`. Closes the PR without merging.
        The companion `delete_branch` call lives next to the rollback
        caller in `api_graduation.py` per the §19.2 'graduation rollback's
        branch cleanup' candidate Slice 8 settles — the §12 hygiene
        sweep would catch the branch eventually, but closing the loop
        on rollback avoids accumulation."""
        await self._gitea.close_pull(org, meta_repo, pr_number)
        _log(
            actor,
            "graduate_pr_close",
            rfc_slug=slug,
            branch_name=head_branch,
            pr_number=pr_number,
            details={"reason": reason},
        )

    # ----- §12 hygiene: branch deletion -----

    async def delete_branch(
        self,
        actor: Actor | None,
        *,
        owner: str,
        repo: str,
        branch: str,
        slug: str | None,
        action_kind: str,
        reason: str,
        bot_login: str | None = None,
    ) -> None:
        """Per §12: the bot deletes a stale branch from Gitea.

        Three callers: the §12 hygiene sweep (90-day boundary on
        meta-repo edit branches), the §10.7 90-day post-merge timer
        for per-RFC PR branches, and the graduation-rollback cleanup
        for `graduate-<slug>-<6hex>` per §19.2 "graduation rollback's
        branch cleanup."

        For the timer paths the caller passes `actor=None`; the audit
        row lands with `actor_user_id=NULL` and `on_behalf_of=bot_login`
        per §15.9's "system-generated events" rule — "the app" in the
        noun slot. For the rollback case the human actor flows through
        the standard `_log` shape.

        Idempotent against the Gitea API — 404 from a prior delete is
        swallowed so a retried sweep doesn't crash.
        """
        try:
            await self._gitea.delete_branch(owner, repo, branch)
        except Exception as exc:
            from .gitea import GiteaError as _GE
            if isinstance(exc, _GE) and exc.status == 404:
                pass
            else:
                raise
        details = {"repo": f"{owner}/{repo}", "reason": reason}
        if actor is None:
            # System actor: write the audit row directly. Fan-out is
            # skipped — `delete_stale_branch` and `delete_post_merge_branch`
            # are intentionally absent from `notify._AUTO_WATCH_ACTIONS`
            # and `_ROUTING`, so no notification fires. The branches
            # being deleted are stale; the population that watched them
            # would be churn-grade noise per §15.4.
            db.conn().execute(
                """
                INSERT INTO actions
                    (actor_user_id, on_behalf_of, action_kind, rfc_slug, branch_name, details)
                VALUES (NULL, ?, ?, ?, ?, ?)
                """,
                (
                    bot_login or "",
                    action_kind,
                    slug,
                    branch,
                    json.dumps(details),
                ),
            )
            return
        _log(
            actor,
            action_kind,
            rfc_slug=slug,
            branch_name=branch,
            details=details,
        )

    # ----- §13.1 claim PRs -----

    async def open_claim_pr(
        self,
        actor: Actor,
        *,
        org: str,
        meta_repo: str,
        slug: str,
        new_file_contents: str,
        prior_sha: str,
    ) -> dict:
        """§13.1: open a PR adding the actor to the entry's `owners:` list.

        Touches only the frontmatter of `rfcs/<slug>.md`. Branch shape is
        `claim/<slug>` — single attempt per super-draft per actor (Gitea
        refuses duplicate branch creation, which is the right behavior:
        if the claim is still open, point the contributor at the existing
        PR rather than opening a second one).
        """
        branch = f"claim/{slug}"
        await self._gitea.create_branch(org, meta_repo, branch, from_branch="main")
        ae = actor.email or f"{actor.gitea_login}@users.noreply"
        commit_subject = f"Claim ownership of {slug} for {actor.gitea_login}"
        commit_message = _stamp_single(commit_subject, actor)
        result = await self._gitea.update_file(
            org, meta_repo, f"rfcs/{slug}.md",
            content=new_file_contents,
            sha=prior_sha,
            message=commit_message,
            branch=branch,
            author_name=actor.display_name, author_email=ae,
        )
        commit_sha = (
            result.get("commit", {}).get("sha")
            or result.get("content", {}).get("sha")
            or ""
        )
        pr_title = f"Claim ownership: {slug}"
        pr_description = (
            f"`{actor.gitea_login}` claims ownership of super-draft `{slug}`.\n\n"
            f"Per §13.1, owners and admins can merge."
        )
        _subject, pr_body = _stamp("", pr_description, actor)
        pr = await self._gitea.create_pull(
            org, meta_repo,
            title=pr_title, body=pr_body, head=branch, base="main",
        )
        _log(
            actor,
            "open_claim_pr",
            rfc_slug=slug,
            branch_name=branch,
            pr_number=pr["number"],
            bot_commit_sha=commit_sha,
            details={"new_owner": actor.gitea_login},
        )
        return pr

    # ----- §22.4c: mark-reviewed (direct main write) -----

    async def mark_entry_reviewed(
        self,
        actor: Actor,
        *,
        org: str,
        meta_repo: str,
        slug: str,
        reviewed_by: str,
        reviewed_at: str,
    ) -> None:
        """Clear §22.4c unreviewed on an active entry by rewriting its
        frontmatter on main. Stamps the commit with the §6.5 On-behalf-of
        trailer and writes an actions-log row, mirroring the graduation
        stamp's bot-write shape."""
        path = f"rfcs/{slug}.md"
        result = await self._gitea.read_file(org, meta_repo, path, ref="main")
        if result is None:
            raise GiteaError(404, f"{path} not found")
        text, sha = result
        e = entry_mod.parse(text)
        e.unreviewed = False
        e.reviewed_at = reviewed_at
        e.reviewed_by = reviewed_by
        commit_message = _stamp_single(f"Mark {slug} reviewed", actor)
        result = await self._gitea.update_file(
            org, meta_repo, path,
            content=entry_mod.serialize(e),
            sha=sha,
            message=commit_message,
            branch="main",
            author_name=actor.display_name,
            author_email=actor.email or f"{actor.gitea_login}@users.noreply",
        )
        commit_sha = (
            result.get("commit", {}).get("sha")
            or result.get("content", {}).get("sha")
            or ""
        )
        _log(
            actor,
            "mark_reviewed",
            rfc_slug=slug,
            bot_commit_sha=commit_sha,
            details={"reviewed_by": reviewed_by, "reviewed_at": reviewed_at},
        )

    # ----- Per-RFC repo: seeding (test/dev fixtures, future graduation) -----

    async def ensure_rfc_repo_seed(
        self,
        actor: Actor,
        *,
        owner: str,
        repo: str,
        slug: str,
        title: str,
        body: str,
    ) -> None:
        """Create the per-RFC repo and seed `RFC.md` on `main` if missing.

        Slice 2 surfaces against per-RFC repos that Slice 5's graduation
        flow will eventually create. Until graduation exists, this is the
        seam test fixtures and ad-hoc dev workflows use to bring an RFC
        repo into existence — the bot stays the only Git writer and the
        seed itself enters the audit log.
        """
        existing = await self._gitea.get_repo(owner, repo)
        if existing is None:
            await self._gitea.create_org_repo(owner, repo, description=f"RFC: {title}")
        # If main has a tip already, leave it alone — the seed is idempotent.
        main = await self._gitea.get_branch(owner, repo, "main")
        if main is not None:
            return
        message = "Seed RFC.md\n\n" + _trailer(actor)
        await self._gitea.create_file(
            owner,
            repo,
            "RFC.md",
            content=body,
            message=message,
            branch="main",
            author_name=actor.display_name,
            author_email=actor.email or f"{actor.gitea_login}@users.noreply",
        )
        _log(
            actor,
            "seed_rfc_repo",
            rfc_slug=slug,
            branch_name="main",
            details={"repo": f"{owner}/{repo}", "title": title},
        )


def _subject_from_reason(reason: str, fallback: str) -> str:
    """One-line commit subject derived from the change's reason.

    Truncated to 72 chars so the Git log scans cleanly. Exact length is
    an implementation detail per §8.6.
    """
    text = (reason or "").strip().split("\n")[0]
    if not text:
        return fallback
    if len(text) > 72:
        return text[:69].rstrip() + "…"
    return text