rfc-app/backend/app/cache.py

"""The §4 metadata cache and its two writers.

Per §4: Gitea is truth. The cache mirrors only what the left pane and
the read surfaces need, and it is rebuildable from Gitea at any time.
Per §4.1: two writers — the webhook handler and the periodic reconciler —
both read from Gitea and write to the cache. User actions never write
to the cache directly; they trigger Git operations through the bot
(`bot.py`), and the resulting webhook (or the next reconciler sweep)
is what updates the cache.

This module provides:
  - `refresh_meta_repo()` — reads rfcs/ on the meta repo and reconciles
    cached_rfcs against what's there. Used by both the webhook handler
    (on meta-repo merge events) and the reconciler.
  - `refresh_meta_pulls()` — reads open meta-repo PRs and reconciles
    cached_prs for pr_kind='idea' and friends. Backs the §7.3
    pending-ideas disclosure.

Per §4.2's "single SQLite file colocated with the FastAPI process," the
cache writes happen on the same process that serves reads; lock
contention is bounded by the small mutation surface (a few hundred
rows at most for v1) and SQLite's WAL mode.
"""
from __future__ import annotations

import asyncio
import json
import logging

from . import db, entry as entry_mod, projects as projects_mod, registry as registry_mod
from .config import Config
from .gitea import Gitea, GiteaError

log = logging.getLogger(__name__)


async def refresh_meta_repo(config: Config, gitea: Gitea) -> None:
    """Re-read rfcs/ on every project's content repo and reconcile cached_rfcs.

    §22 (Plan B): a deployment has N projects (§22.1), each with its own
    content_repo (§22.3). Mirror each into cached_rfcs stamped with that
    project's id, so a second project's corpus renders under /p/<id>/. Idempotent;
    safe on every content-repo webhook and reconciler sweep.
    """
    org = config.gitea_org
    rows = db.conn().execute(
        "SELECT id, content_repo FROM projects WHERE content_repo IS NOT NULL AND content_repo != ''"
    ).fetchall()
    if not rows:
        log.warning("refresh_meta_repo: no projects with a content_repo yet; skipping")
        return
    for prow in rows:
        await _refresh_project_corpus(org, prow["id"], prow["content_repo"], gitea)


async def _refresh_project_corpus(org: str, project_id: str, repo: str, gitea: Gitea) -> None:
    try:
        files = await gitea.list_dir(org, repo, "rfcs", ref="main")
    except GiteaError as e:
        log.warning("refresh_meta_repo: project %s: cannot list rfcs/: %s", project_id, e)
        return

    seen_slugs: set[str] = set()
    for f in files:
        if f.get("type") != "file" or not f.get("name", "").endswith(".md"):
            continue
        result = await gitea.read_file(org, repo, f["path"], ref="main")
        if not result:
            continue
        text, sha = result
        try:
            entry = entry_mod.parse(text)
        except Exception as parse_err:
            log.warning("refresh_meta_repo: %s: skipping %s: %s", project_id, f["path"], parse_err)
            continue
        if not entry.slug:
            log.warning("refresh_meta_repo: %s: skipping %s: missing slug", project_id, f["path"])
            continue
        seen_slugs.add(entry.slug)
        _upsert_cached_rfc(entry, body_sha=sha, project_id=project_id)

    # Entries removed from a project's rfcs/ — the spec keeps withdrawn entries
    # as historical record (§3), so this fires only for out-of-band deletes;
    # leave the row, scoped to this project, for reconciler attention.
    existing = {
        row["slug"]
        for row in db.conn().execute(
            "SELECT slug FROM cached_rfcs WHERE project_id = ?", (project_id,)
        )
    }
    for missing in existing - seen_slugs:
        log.info("refresh_meta_repo: %s/%s no longer in rfcs/ — leaving cache row", project_id, missing)


def _upsert_cached_rfc(entry: entry_mod.Entry, body_sha: str, project_id: str = "default") -> None:
    # §6.6: models_json stays NULL when the frontmatter key is absent
    # (inherit operator universe) and '[]' for the explicit opt-out.
    models_json = json.dumps(entry.models) if entry.models is not None else None
    # §6.7: funder_login mirrors the optional `funder:` frontmatter
    # field. NULL means absent — operator credentials are used.
    funder_login = entry.funder or None
    db.conn().execute(
        """
        INSERT INTO cached_rfcs
            (slug, title, state, rfc_id, repo, proposed_by, proposed_at,
             graduated_at, graduated_by, owners_json, arbiters_json, tags_json,
             models_json, funder_login, body, body_sha,
             unreviewed, reviewed_at, reviewed_by, project_id,
             last_entry_commit_at, updated_at)
        VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, datetime('now'), datetime('now'))
        ON CONFLICT(project_id, slug) DO UPDATE SET
            title = excluded.title,
            state = excluded.state,
            rfc_id = excluded.rfc_id,
            repo = excluded.repo,
            proposed_by = excluded.proposed_by,
            proposed_at = excluded.proposed_at,
            graduated_at = excluded.graduated_at,
            graduated_by = excluded.graduated_by,
            owners_json = excluded.owners_json,
            arbiters_json = excluded.arbiters_json,
            tags_json = excluded.tags_json,
            models_json = excluded.models_json,
            funder_login = excluded.funder_login,
            body = excluded.body,
            body_sha = excluded.body_sha,
            unreviewed = excluded.unreviewed,
            reviewed_at = excluded.reviewed_at,
            reviewed_by = excluded.reviewed_by,
            last_entry_commit_at = datetime('now'),
            updated_at = datetime('now')
        """,
        (
            entry.slug,
            entry.title,
            entry.state,
            entry.id,
            entry.repo,
            entry.proposed_by,
            entry.proposed_at,
            entry.graduated_at,
            entry.graduated_by,
            json.dumps(entry.owners),
            json.dumps(entry.arbiters),
            json.dumps(entry.tags),
            models_json,
            funder_login,
            entry.body,
            body_sha,
            1 if entry.unreviewed else 0,
            entry.reviewed_at,
            entry.reviewed_by,
            project_id,
        ),
    )


async def refresh_rfc_repo(config: Config, gitea: Gitea, slug: str) -> None:
    """Mirror an active RFC's per-RFC repo into the cache.

    Reads `RFC.md` on main into `cached_rfcs.body` (per §4 #3), lists
    branches into `cached_branches`, and lists open PRs into
    `cached_prs` with `pr_kind='rfc_branch'`. Per §4.1 this runs in two
    places: a webhook arrival for events on the per-RFC repo, and the
    reconciler sweep.
    """
    row = db.conn().execute(
        "SELECT repo, state FROM cached_rfcs WHERE slug = ?", (slug,)
    ).fetchone()
    if not row or not row["repo"] or row["state"] != "active":
        return
    if "/" not in row["repo"]:
        log.warning("refresh_rfc_repo: %s has malformed repo %r", slug, row["repo"])
        return
    owner, repo = row["repo"].split("/", 1)

    # Body on main — populates the discuss-mode default surface per §8.2.
    try:
        result = await gitea.read_file(owner, repo, "RFC.md", ref="main")
    except GiteaError as e:
        log.warning("refresh_rfc_repo(%s): read_file failed: %s", slug, e)
        result = None
    if result is not None:
        text, sha = result
        db.conn().execute(
            """
            UPDATE cached_rfcs
               SET body = ?, body_sha = ?, last_main_commit_at = datetime('now'),
                   updated_at = datetime('now')
             WHERE slug = ?
            """,
            (text, sha, slug),
        )

    # Branches — every branch the bot knows about per §11.5 / §12.
    try:
        branches = await gitea.list_branches(owner, repo)
    except GiteaError as e:
        log.warning("refresh_rfc_repo(%s): list_branches failed: %s", slug, e)
        branches = []
    seen_branches: set[str] = set()
    for b in branches:
        name = b.get("name") or ""
        if not name:
            continue
        seen_branches.add(name)
        head_sha = (b.get("commit") or {}).get("id") or ""
        last_commit_at = (b.get("commit") or {}).get("timestamp")
        db.conn().execute(
            """
            INSERT INTO cached_branches (rfc_slug, branch_name, head_sha, state, last_commit_at)
            VALUES (?, ?, ?, 'open', ?)
            ON CONFLICT(project_id, rfc_slug, branch_name) DO UPDATE SET
                head_sha = excluded.head_sha,
                state = CASE WHEN cached_branches.state = 'closed' THEN 'closed' ELSE 'open' END,
                last_commit_at = excluded.last_commit_at
            """,
            (slug, name, head_sha, last_commit_at),
        )
    # Mark previously-known branches that disappeared as deleted, keeping
    # the row per §11.5 ("branch removed from Gitea, row remains").
    existing = {
        r["branch_name"]
        for r in db.conn().execute(
            "SELECT branch_name FROM cached_branches WHERE rfc_slug = ? AND state != 'deleted'",
            (slug,),
        )
    }
    for missing in existing - seen_branches:
        db.conn().execute(
            "UPDATE cached_branches SET state = 'deleted' WHERE rfc_slug = ? AND branch_name = ?",
            (slug, missing),
        )

    # PRs on the per-RFC repo (pr_kind = 'rfc_branch'). Slice 3 owns the
    # full PR surface; we mirror metadata here so the §8.1 breadcrumb
    # dropdown's "1 PR" count is honest from Slice 2 onward.
    repo_full = f"{owner}/{repo}"
    bot_login = config.gitea_bot_user
    try:
        open_pulls = await gitea.list_pulls(owner, repo, state="open")
        closed_pulls = await gitea.list_pulls(owner, repo, state="closed")
    except GiteaError as e:
        log.warning("refresh_rfc_repo(%s): list_pulls failed: %s", slug, e)
        open_pulls, closed_pulls = [], []
    for pull in open_pulls + closed_pulls:
        head_branch = pull.get("head", {}).get("ref", "")
        # Same deleted-branch recovery as refresh_meta_pulls: a merged-and-
        # deleted PR's `head.ref` collapses to `refs/pull/<N>/head`. Here
        # the slug is known (param), so state still updates correctly and
        # no ghost forms — but blindly storing the sentinel would clobber
        # the real branch name api_prs.py relies on as a fallback ref when
        # the merge commit is gone. Recover it from the stored row.
        if not head_branch or head_branch.startswith("refs/pull/"):
            prior = db.conn().execute(
                "SELECT head_branch FROM cached_prs WHERE repo = ? AND pr_number = ?",
                (repo_full, pull["number"]),
            ).fetchone()
            if prior and prior["head_branch"]:
                head_branch = prior["head_branch"]
        state = _state_from_pull(pull)
        gitea_opener = (pull.get("user") or {}).get("login") or ""
        opened_by = _resolve_actor(
            gitea_opener,
            bot_login,
            slug,
            pull["number"],
            pull.get("body") or "",
        )
        # §10.8: distinguish "user withdrew" from "Gitea closed for any
        # other reason." The bot's withdraw action lands in the actions
        # log; if we see it, surface state='withdrawn'.
        if state == "closed":
            withdrew = db.conn().execute(
                """
                SELECT 1 FROM actions
                 WHERE action_kind = 'withdraw_branch_pr'
                   AND rfc_slug = ? AND pr_number = ? LIMIT 1
                """,
                (slug, pull["number"]),
            ).fetchone()
            if withdrew:
                state = "withdrawn"
        merge_commit_sha = pull.get("merge_commit_sha")
        db.conn().execute(
            """
            INSERT INTO cached_prs
                (rfc_slug, pr_kind, repo, pr_number, title, description, state,
                 opened_by, opened_at, merged_at, closed_at,
                 head_branch, base_branch, head_sha, merge_commit_sha)
            VALUES (?, 'rfc_branch', ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            ON CONFLICT(repo, pr_number) DO UPDATE SET
                title = excluded.title,
                description = excluded.description,
                state = excluded.state,
                opened_by = excluded.opened_by,
                merged_at = excluded.merged_at,
                closed_at = excluded.closed_at,
                head_sha = excluded.head_sha,
                merge_commit_sha = COALESCE(excluded.merge_commit_sha, cached_prs.merge_commit_sha)
            """,
            (
                slug,
                repo_full,
                pull["number"],
                pull.get("title") or "",
                pull.get("body") or "",
                state,
                opened_by,
                pull.get("created_at"),
                pull.get("merged_at"),
                pull.get("closed_at"),
                head_branch,
                (pull.get("base") or {}).get("ref") or "main",
                (pull.get("head") or {}).get("sha"),
                merge_commit_sha,
            ),
        )
        # §10.9: an explicit `Supersedes: #N` trailer on a merged PR's
        # body bumps the predecessor's state to closed and records the
        # supersession. The cache propagates this whether the merge came
        # via webhook or reconciler.
        if state == "merged":
            superseded = _parse_supersedes(pull.get("body") or "")
            if superseded:
                db.conn().execute(
                    """
                    UPDATE cached_prs
                       SET state = 'closed',
                           superseded_by_pr_number = ?,
                           closed_at = COALESCE(closed_at, datetime('now'))
                     WHERE repo = ? AND pr_number = ? AND state = 'open'
                    """,
                    (pull["number"], repo_full, superseded),
                )


async def refresh_meta_branches(config: Config, gitea: Gitea) -> None:
    """Mirror the meta repo's branches into `cached_branches` for super-draft
    edit branches, plus a per-slug `main` row that records the meta-repo
    main's tip sha so the §10.1 has-commits-ahead check works uniformly
    across active and super-draft surfaces.

    Per the §5 super-draft scoping note, super-draft edits are branches on
    the meta repo. The naming Slice 4 picked is `edit-<slug>-<6hex>` —
    structurally `edit/<slug>/<auto-name>` per §9.5, with dashes in place
    of slashes per the §19.2 path-routing candidate.
    """
    org = config.gitea_org
    repo = projects_mod.default_content_repo(config)
    if not repo:
        log.warning("refresh_meta_branches: default project has no content_repo yet; skipping")
        return
    try:
        branches = await gitea.list_branches(org, repo)
    except GiteaError as e:
        log.warning("refresh_meta_branches: %s", e)
        return

    meta_main_sha = ""
    meta_main_ts = None
    edit_keys_seen: set[tuple[str, str]] = set()
    for b in branches:
        name = b.get("name") or ""
        head_sha = (b.get("commit") or {}).get("id") or ""
        last_commit_at = (b.get("commit") or {}).get("timestamp")
        if name == "main":
            meta_main_sha = head_sha
            meta_main_ts = last_commit_at
            continue
        slug = _slug_from_branch_name(name)
        if not slug:
            continue
        rfc = db.conn().execute(
            "SELECT state, repo FROM cached_rfcs WHERE slug = ?", (slug,)
        ).fetchone()
        # Meta-only topology (§1): edit branches live on the meta repo for
        # every meta-resident entry — super-drafts and active RFCs alike
        # (active RFCs are graduated in place and keep editing here, §13).
        # A legacy per-RFC repo (repo set) is the only thing excluded.
        if not rfc or rfc["repo"] or rfc["state"] not in ("super-draft", "active"):
            continue
        edit_keys_seen.add((slug, name))
        db.conn().execute(
            """
            INSERT INTO cached_branches (rfc_slug, branch_name, head_sha, state, last_commit_at)
            VALUES (?, ?, ?, 'open', ?)
            ON CONFLICT(project_id, rfc_slug, branch_name) DO UPDATE SET
                head_sha = excluded.head_sha,
                state = CASE WHEN cached_branches.state = 'closed' THEN 'closed' ELSE 'open' END,
                last_commit_at = excluded.last_commit_at
            """,
            (slug, name, head_sha, last_commit_at),
        )

    # Synthesize a per-slug `main` row for every super-draft entry, so the
    # §10.1 has-commits-ahead check in api_prs.py works uniformly. The
    # head_sha is the meta-repo main's tip — every super-draft edit branch
    # diverges from this single point.
    if meta_main_sha:
        super_drafts = db.conn().execute(
            "SELECT slug FROM cached_rfcs "
            "WHERE repo IS NULL AND state IN ('super-draft', 'active')"
        ).fetchall()
        for r in super_drafts:
            db.conn().execute(
                """
                INSERT INTO cached_branches (rfc_slug, branch_name, head_sha, state, last_commit_at)
                VALUES (?, 'main', ?, 'open', ?)
                ON CONFLICT(project_id, rfc_slug, branch_name) DO UPDATE SET
                    head_sha = excluded.head_sha,
                    last_commit_at = excluded.last_commit_at
                """,
                (r["slug"], meta_main_sha, meta_main_ts),
            )

    # Mark previously-known edit branches that disappeared as deleted per
    # §11.5 / §12. Keep the row so chat history survives the branch's
    # deletion in Gitea.
    known = db.conn().execute(
        """
        SELECT b.rfc_slug, b.branch_name
          FROM cached_branches b
          JOIN cached_rfcs r ON r.slug = b.rfc_slug
         WHERE r.repo IS NULL
           AND r.state IN ('super-draft', 'active')
           AND b.state != 'deleted'
           AND b.branch_name != 'main'
        """
    ).fetchall()
    for k in known:
        if (k["rfc_slug"], k["branch_name"]) not in edit_keys_seen:
            db.conn().execute(
                "UPDATE cached_branches SET state = 'deleted' WHERE rfc_slug = ? AND branch_name = ?",
                (k["rfc_slug"], k["branch_name"]),
            )


def _slug_from_branch_name(name: str) -> str | None:
    """Mirror of `_slug_from_head_branch` for branch-only inputs (no PR
    body to consult)."""
    if name.startswith("edit-"):
        body = name[len("edit-") :]
        if "-" in body:
            slug, _hex = body.rsplit("-", 1)
            return slug or None
    if name.startswith("edit/"):
        parts = name.split("/", 2)
        if len(parts) >= 2:
            return parts[1]
    return None


async def refresh_meta_pulls(config: Config, gitea: Gitea) -> None:
    """Reconcile open meta-repo PRs into cached_prs.

    For Slice 1 we care about pr_kind='idea' (proposing a new entry).
    Other meta-repo PR kinds (body edits, metadata edits, claims) will
    be wired in their respective slices.

    `opened_by` is the **underlying actor**, not the bot login Gitea
    reports — per §15.9's framing for notifications and per §6.5's
    On-behalf-of accountability shape. We recover the actor by joining
    against the `actions` audit log; if no row matches (cache rebuilt
    from scratch on a deployment that pre-dates the actions log, or a
    pull we did not author), we fall back to parsing the
    `On-behalf-of:` trailer from the PR body, then to the raw Gitea
    login as last resort.
    """
    org = config.gitea_org
    bot_login = config.gitea_bot_user
    rows = db.conn().execute(
        "SELECT id, content_repo FROM projects WHERE content_repo IS NOT NULL AND content_repo != ''"
    ).fetchall()
    if not rows:
        log.warning("refresh_meta_pulls: no projects with a content_repo yet; skipping")
        return
    for prow in rows:
        await _refresh_project_pulls(org, prow["id"], prow["content_repo"], gitea, bot_login)


async def _refresh_project_pulls(
    org: str, project_id: str, repo: str, gitea: Gitea, bot_login: str
) -> None:
    repo_full = f"{org}/{repo}"
    try:
        open_pulls = await gitea.list_pulls(org, repo, state="open")
        closed_pulls = await gitea.list_pulls(org, repo, state="closed")
    except GiteaError as e:
        log.warning("refresh_meta_pulls: project %s: %s", project_id, e)
        return

    for pull in open_pulls + closed_pulls:
        head_branch = pull.get("head", {}).get("ref", "")
        # A merged-and-deleted PR's branch is no longer reported by Gitea
        # as its real name — the `head.ref` collapses to the synthetic
        # `refs/pull/<N>/head` sentinel (or empty). The slug + kind both
        # derive from the branch name, so a deleted branch would parse to
        # slug=None and the row would be skipped forever, freezing the
        # cached_prs row at its last-seen `state='open'` — a permanent
        # ghost "pending idea" for an entry that has actually merged
        # (caught when the operator authoring lane in ROADMAP #35 merged
        # an idea PR with the branch deleted; the web UX leaves branches
        # in place so it never tripped this). Recover the original branch
        # from the row we already stored when the PR was open — that row
        # retains the real `head_branch` (migration 002).
        if not head_branch or head_branch.startswith("refs/pull/"):
            prior = db.conn().execute(
                "SELECT head_branch FROM cached_prs WHERE repo = ? AND pr_number = ?",
                (repo_full, pull["number"]),
            ).fetchone()
            if prior and prior["head_branch"]:
                head_branch = prior["head_branch"]
        slug = _slug_from_head_branch(head_branch)
        if slug is None:
            continue
        pr_kind = _kind_from_branch(head_branch)
        state = _state_from_pull(pull)
        gitea_opener = (pull.get("user") or {}).get("login") or ""
        opened_by = _resolve_actor(
            gitea_opener,
            bot_login,
            slug,
            pull["number"],
            pull.get("body") or "",
        )
        # §10.8 / Slice 4: a closed body-edit PR may have been withdrawn
        # by the contributor. Distinguish from a generic Gitea close via
        # the audit log — same shape api_prs.py uses for rfc_branch PRs.
        if state == "closed" and pr_kind == "meta_body_edit":
            withdrew = db.conn().execute(
                """
                SELECT 1 FROM actions
                 WHERE action_kind = 'withdraw_branch_pr'
                   AND rfc_slug = ? AND pr_number = ? LIMIT 1
                """,
                (slug, pull["number"]),
            ).fetchone()
            if withdrew:
                state = "withdrawn"
        merge_commit_sha = pull.get("merge_commit_sha")
        db.conn().execute(
            """
            INSERT INTO cached_prs
                (rfc_slug, pr_kind, repo, pr_number, title, description, state,
                 opened_by, opened_at, merged_at, closed_at,
                 head_branch, base_branch, head_sha, merge_commit_sha, project_id)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            ON CONFLICT(repo, pr_number) DO UPDATE SET
                title = excluded.title,
                description = excluded.description,
                state = excluded.state,
                opened_by = excluded.opened_by,
                merged_at = excluded.merged_at,
                closed_at = excluded.closed_at,
                head_sha = excluded.head_sha,
                merge_commit_sha = COALESCE(excluded.merge_commit_sha, cached_prs.merge_commit_sha)
            """,
            (
                slug,
                pr_kind,
                repo_full,
                pull["number"],
                pull.get("title") or "",
                pull.get("body") or "",
                state,
                opened_by,
                pull.get("created_at"),
                pull.get("merged_at"),
                pull.get("closed_at"),
                head_branch,
                (pull.get("base") or {}).get("ref") or "main",
                (pull.get("head") or {}).get("sha"),
                merge_commit_sha,
                project_id,
            ),
        )


_TRAILER_RE = None


def _resolve_actor(gitea_opener: str, bot_login: str, slug: str, pr_number: int, body: str) -> str:
    """Best effort: collapse the bot's authorship to the underlying actor."""
    if gitea_opener and gitea_opener != bot_login:
        return gitea_opener
    # Prefer the audit log.
    row = db.conn().execute(
        """
        SELECT on_behalf_of FROM actions
         WHERE action_kind IN ('propose_rfc', 'open_body_edit_pr', 'open_branch_pr', 'open_claim_pr', 'open_metadata_pr')
           AND rfc_slug = ? AND pr_number = ?
         ORDER BY id LIMIT 1
        """,
        (slug, pr_number),
    ).fetchone()
    if row and row["on_behalf_of"]:
        return row["on_behalf_of"]
    # Fall back to parsing the On-behalf-of trailer.
    import re as _re
    global _TRAILER_RE
    if _TRAILER_RE is None:
        _TRAILER_RE = _re.compile(r"On-behalf-of:\s+.*?<([^>]+)>", _re.MULTILINE)
    m = _TRAILER_RE.search(body)
    if m:
        return m.group(1)
    return gitea_opener or bot_login


def _slug_from_head_branch(head_branch: str) -> str | None:
    if head_branch.startswith("propose/"):
        return head_branch[len("propose/") :]
    if head_branch.startswith("edit/"):
        parts = head_branch.split("/", 2)
        if len(parts) >= 2:
            return parts[1]
    if head_branch.startswith("edit-"):
        # §9.5 names the structural shape `edit/<slug>/<auto-name>`, but
        # FastAPI's default {branch} path-segment matcher refuses slashes
        # (the §19.2 routing candidate). Slice 4 picks the same dash-
        # separated workaround Slice 2 used for promote-to-branch:
        # `edit-<slug>-<6hex>`. The slug is the middle; the final
        # dash-segment is a 6-hex suffix.
        body = head_branch[len("edit-") :]
        if "-" in body:
            slug, _hex = body.rsplit("-", 1)
            return slug or None
    if head_branch.startswith("claim/"):
        return head_branch[len("claim/") :]
    if head_branch.startswith("metadata/"):
        return head_branch[len("metadata/") :]
    if head_branch.startswith("metadata-"):
        # §9.5 metadata-pane PRs use the same dash-separated branch shape
        # as edit branches, for the same routing reason.
        body = head_branch[len("metadata-") :]
        if "-" in body:
            slug, _hex = body.rsplit("-", 1)
            return slug or None
    return None


def _kind_from_branch(head_branch: str) -> str:
    if head_branch.startswith("propose/"):
        return "idea"
    if head_branch.startswith("edit/") or head_branch.startswith("edit-"):
        return "meta_body_edit"
    if head_branch.startswith("claim/"):
        return "meta_claim"
    if head_branch.startswith("metadata/") or head_branch.startswith("metadata-"):
        return "meta_metadata"
    return "idea"  # fallback


_SUPERSEDES_RE = None


def _parse_supersedes(body: str) -> int | None:
    """Parse a `Supersedes: #N` trailer from a PR body per §10.9."""
    import re as _re
    global _SUPERSEDES_RE
    if _SUPERSEDES_RE is None:
        _SUPERSEDES_RE = _re.compile(r"^Supersedes:\s*#(\d+)", _re.MULTILINE)
    m = _SUPERSEDES_RE.search(body or "")
    return int(m.group(1)) if m else None


def _state_from_pull(pull: dict) -> str:
    if pull.get("merged"):
        return "merged"
    if pull.get("state") == "closed":
        return "closed"
    return "open"


# ----- Reconciler -----

class Reconciler:
    """Per §4.1: periodic safety-net sweep.

    Runs in the background, every five minutes by default. Catches up
    on any webhook the bot missed (downtime, network failure, Gitea
    flake). If the cache is corrupted, the reconciler rebuilds from
    scratch — that's the contract.
    """

    def __init__(self, config: Config, gitea: Gitea, interval_seconds: int = 300):
        self._config = config
        self._gitea = gitea
        self._interval = interval_seconds
        self._task: asyncio.Task | None = None
        self._stop = asyncio.Event()

    async def _loop(self) -> None:
        # One sweep at startup, then on the interval. The startup sweep
        # is what brings a fresh cache to life on first boot.
        await self.sweep()
        while not self._stop.is_set():
            try:
                await asyncio.wait_for(self._stop.wait(), timeout=self._interval)
            except asyncio.TimeoutError:
                pass
            if self._stop.is_set():
                break
            await self.sweep()

    async def sweep(self) -> None:
        log.info("reconciler: starting sweep")
        try:
            try:
                await registry_mod.refresh_registry(self._config, self._gitea)
            except Exception:
                log.exception("reconciler: registry refresh failed; keeping last-good projects")
            await refresh_meta_repo(self._config, self._gitea)
            await refresh_meta_branches(self._config, self._gitea)
            await refresh_meta_pulls(self._config, self._gitea)
            # Per-RFC repos: refresh each active entry. Meta-repo refresh
            # must come first so newly-graduated entries land in
            # cached_rfcs before we try to reach their per-RFC repos.
            active = [
                r["slug"]
                for r in db.conn().execute(
                    "SELECT slug FROM cached_rfcs WHERE state = 'active' AND repo IS NOT NULL"
                )
            ]
            for slug in active:
                await refresh_rfc_repo(self._config, self._gitea, slug)
        except Exception:
            log.exception("reconciler: sweep failed")
        else:
            log.info("reconciler: sweep complete")

    def start(self) -> None:
        if self._task is None:
            self._task = asyncio.create_task(self._loop())

    async def stop(self) -> None:
        self._stop.set()
        if self._task is not None:
            await self._task
            self._task = None