rfc-app/backend/app/rfc_links.py

"""Roadmap #28 — scan submitted prose for RFC-shaped references.

The scanner splits a plain-text PR description / comment body into a list
of *segments* the frontend renders: plain-text runs interleaved with
typed link segments. The backend never emits HTML — the frontend maps
each segment onto a React node — so the surface is XSS-safe by
construction and independent of any HTML-sanitization layer.

Three buckets, one scan (Parts 1–3):

  * ``{"type": "rfc", ...}``          — Part 1. The term matches an
    **accepted** (``state='active'``) RFC; renders as a link to it.
  * ``{"type": "rfc-pending", ...}``  — Part 3. The term matches a
    **pending** RFC — a super-draft (``state='super-draft'``: accepted
    as an idea but not yet graduated to an active RFC) — which has an
    owner and a contribution surface. Renders as an "ask to contribute"
    affordance carrying the owner's display name.
  * ``{"type": "rfc-candidate", ...}`` — Part 2. The term is a
    strong-candidate that does **not** yet have a defining RFC. Renders
    (for a viewer with create rights) as a "create RFC for '<term>'"
    affordance that pre-fills the propose flow.

Precedence at any position is active > pending > candidate, then
longest-match-first — an active link always wins over a contribute offer
which always wins over a create offer for the same span.

**Read-time enrichment, not submit-time persistence** (unchanged from
Part 1): drafts are never scanned, only submitted content on the read
paths, so links/offers track the *live* corpus. The active-RFC corpus,
super-draft corpus, and tag taxonomy are all small and cache-resident,
so building the index and scanning a ≤20k-char body per read is cheap.

**Matching stays conservative by design.** A reference links/offers only
when it is unlikely to be coincidental:

  * ``rfc_id`` tokens (e.g. ``RFC-0001``) — inherently specific.
  * Multi-word titles (containing whitespace, e.g. ``Open Human Model``).
  * Hyphenated slugs (containing ``-``, e.g. ``open-human-model``).

Single common-word titles/slugs are deliberately NOT matched — they
would turn every prose occurrence into an affordance.

**Part 2 candidate heuristic.** A candidate term is a **multi-word tag**
from the #27 tag taxonomy (the de-facto set of tags the corpus already
carries) that has no defining RFC (no active or super-draft RFC whose
slug or title is that term). Multi-word is the same false-positive guard
the title rule uses: a single common tag word (``identity``) would be
far too noisy. Broader candidate detection — capitalized multi-word
phrases mined from the text, terms repeated across recently-touched PRs,
or the #27 Haiku (``ANTHROPIC_API_KEY``) pathway — is a sanctioned but
deferred extension; the conservative tag-taxonomy heuristic is chosen
here to match Part 1's false-positive-averse philosophy.
"""
from __future__ import annotations

import json
import re
from typing import Any, Iterable, NamedTuple


class Term(NamedTuple):
    """One match key plus what to emit when it hits.

    ``key`` is the lowercase span to match (word-boundary, longest-first).
    ``kind`` is ``'active' | 'pending' | 'candidate'`` and selects the
    emitted segment shape. ``slug``/``title`` carry the target RFC (active
    + pending); ``owner`` is the pending RFC's owner display name;
    ``term`` is the candidate's canonical display spelling.
    """

    key: str
    kind: str = "active"
    slug: str = ""
    title: str = ""
    owner: str = ""
    term: str = ""


# Lower number = higher precedence when two keys of equal length match at
# the same position. A real link beats a contribute offer beats a create
# offer.
_KIND_PRIORITY = {"active": 0, "pending": 1, "candidate": 2}


def _coerce(t: Term | tuple) -> Term:
    """Accept the legacy ``(key, slug, title)`` 3-tuple (treated as an
    active term) alongside :class:`Term`, so direct unit-test callers and
    older call sites keep working."""
    if isinstance(t, Term):
        return t
    key, slug, title = t  # legacy active 3-tuple
    return Term(key=key, kind="active", slug=slug, title=title)


def _is_word_char(c: str) -> bool:
    """Word-boundary test. Hyphen and underscore count as word chars so a
    match can't begin or end in the middle of a kebab/snake token."""
    return c.isalnum() or c in ("-", "_")


def _emit(term: Term, label: str) -> dict[str, Any]:
    """The segment dict for a matched ``term``; ``label`` preserves source
    casing."""
    if term.kind == "pending":
        return {
            "type": "rfc-pending",
            "slug": term.slug,
            "label": label,
            "title": term.title,
            "owner": term.owner,
        }
    if term.kind == "candidate":
        return {"type": "rfc-candidate", "label": label, "term": term.term}
    return {"type": "rfc", "slug": term.slug, "label": label, "title": term.title}


def segment_text(text: str | None, terms: Iterable[Term | tuple]) -> list[dict[str, Any]]:
    """Split ``text`` into text / link segments against ``terms``.

    ``terms`` are :class:`Term` objects (or legacy ``(key, slug, title)``
    active 3-tuples). Matching is case-insensitive, respects word
    boundaries on both ends, and prefers the longest key — then higher
    :data:`_KIND_PRIORITY` — at any position.

    Always returns at least one segment; for empty/None input that is a
    single empty text segment, so callers can render uniformly.
    """
    ordered = sorted(
        (_coerce(t) for t in terms),
        key=lambda t: (-len(t.key), _KIND_PRIORITY.get(t.kind, 9)),
    )
    if not text:
        return [{"type": "text", "text": text or ""}]

    out: list[dict[str, Any]] = []
    buf: list[str] = []
    low = text.lower()
    n = len(text)
    i = 0
    while i < n:
        match: tuple[Term, int] | None = None
        for term in ordered:
            klen = len(term.key)
            if klen == 0 or not low.startswith(term.key, i):
                continue
            before = text[i - 1] if i > 0 else ""
            after = text[i + klen] if i + klen < n else ""
            if _is_word_char(before) or _is_word_char(after):
                continue
            match = (term, klen)
            break
        if match is not None:
            term, klen = match
            if buf:
                out.append({"type": "text", "text": "".join(buf)})
                buf = []
            out.append(_emit(term, text[i:i + klen]))
            i += klen
        else:
            buf.append(text[i])
            i += 1
    if buf:
        out.append({"type": "text", "text": "".join(buf)})
    return out


def _keys_for(slug: str, title: str, rfc_id: str | None) -> Iterable[str]:
    """The match keys an RFC contributes. See the module docstring for why
    each gate exists (conservative, false-positive-averse)."""
    if rfc_id:
        rid = rfc_id.strip()
        if len(rid) >= 2:
            yield rid.lower()
    if title:
        t = title.strip()
        # Multi-word titles only — a single common word is too noisy.
        if len(t) >= 2 and (" " in t or "\t" in t):
            yield t.lower()
    if slug:
        s = slug.strip()
        # Hyphenated slugs only — a single-token slug is a bare word.
        if len(s) >= 2 and "-" in s:
            yield s.lower()


def _slugify(term: str) -> str:
    """Deterministic kebab-case — mirrors the propose modal's slugify so a
    tag's would-be slug compares correctly against existing RFC slugs."""
    return re.sub(r"-+$", "", re.sub(r"^-+", "", re.sub(r"[^a-z0-9]+", "-", term.lower().strip())))


class LinkIndex:
    """A reusable term index built once per request and applied to many
    bodies (a PR's description plus every comment on it)."""

    def __init__(self, terms: Iterable[Term | tuple]):
        # Coerce + order once; segment_text re-sorts defensively but a
        # pre-sorted list keeps the per-body cost to the scan itself.
        self._terms: list[Term] = sorted(
            (_coerce(t) for t in terms),
            key=lambda t: (-len(t.key), _KIND_PRIORITY.get(t.kind, 9)),
        )

    def __bool__(self) -> bool:
        return bool(self._terms)

    def segment(self, text: str | None) -> list[dict[str, Any]]:
        return segment_text(text, self._terms)


def _owner_display(conn, owners_json: str | None, proposed_by: str | None) -> str:
    """The display name to show for a pending RFC's owner. First entry of
    ``owners_json`` resolved to its user row's display name, falling back
    to the bare login, then ``proposed_by``, then a neutral noun."""
    login = None
    try:
        owners = json.loads(owners_json or "[]")
        if isinstance(owners, list):
            login = next((o for o in owners if isinstance(o, str) and o.strip()), None)
    except (ValueError, TypeError):
        login = None
    if login:
        row = conn.execute(
            "SELECT display_name FROM users WHERE gitea_login = ?", (login,)
        ).fetchone()
        if row and row["display_name"]:
            return row["display_name"]
        return login
    return (proposed_by or "").strip() or "the proposer"


def _tag_universe(conn) -> list[str]:
    """Distinct tags across the cached corpus (the #27 de-facto taxonomy),
    preserving original spelling; case-deduped."""
    rows = conn.execute("SELECT tags_json FROM cached_rfcs").fetchall()
    out: list[str] = []
    seen: set[str] = set()
    for r in rows:
        try:
            tags = json.loads(r["tags_json"] or "[]")
        except (ValueError, TypeError):
            continue
        if not isinstance(tags, list):
            continue
        for t in tags:
            if not isinstance(t, str):
                continue
            tag = t.strip()
            low = tag.lower()
            if tag and low not in seen:
                seen.add(low)
                out.append(tag)
    return out


def build_index(
    conn,
    *,
    exclude_slug: str | None = None,
    include_pending: bool = True,
    include_candidates: bool = True,
) -> LinkIndex:
    """Build a :class:`LinkIndex` over the three buckets.

    ``exclude_slug`` drops the RFC the surrounding surface is itself scoped
    to, so an RFC's own title/id/slug don't self-link (or self-offer)
    inside its own PR or discussion. Precedence is enforced by insertion
    order — active keys are added first and a later bucket never overrides
    an already-claimed key.
    """
    terms: list[Term] = []
    seen: set[str] = set()

    def add(key: str, term: Term) -> None:
        if key in seen:
            return
        seen.add(key)
        terms.append(term)

    # --- Part 1: accepted (active) RFCs. ORDER BY slug makes key
    # de-duplication deterministic when two RFCs would contribute the
    # same key (first slug wins). ---
    active_rows = conn.execute(
        "SELECT slug, title, rfc_id FROM cached_rfcs WHERE state = 'active' ORDER BY slug"
    ).fetchall()
    # Track every slug + title that *has* a defining RFC, so Part 2 never
    # offers to create one that already exists (active or pending).
    defined_slugs: set[str] = set()
    defined_titles: set[str] = set()
    for r in active_rows:
        slug = r["slug"]
        defined_slugs.add((slug or "").lower())
        defined_titles.add((r["title"] or "").strip().lower())
        if exclude_slug is not None and slug == exclude_slug:
            continue
        title = r["title"] or ""
        rfc_id = r["rfc_id"] if "rfc_id" in r.keys() else None
        for key in _keys_for(slug, title, rfc_id):
            add(key, Term(key=key, kind="active", slug=slug, title=title))

    # --- Part 3: pending (super-draft) RFCs. ---
    pending_rows = conn.execute(
        """
        SELECT slug, title, rfc_id, owners_json, proposed_by
          FROM cached_rfcs WHERE state = 'super-draft' ORDER BY slug
        """
    ).fetchall()
    for r in pending_rows:
        slug = r["slug"]
        defined_slugs.add((slug or "").lower())
        defined_titles.add((r["title"] or "").strip().lower())
        if not include_pending:
            continue
        if exclude_slug is not None and slug == exclude_slug:
            continue
        title = r["title"] or ""
        rfc_id = r["rfc_id"] if "rfc_id" in r.keys() else None
        owner = _owner_display(conn, r["owners_json"], r["proposed_by"])
        for key in _keys_for(slug, title, rfc_id):
            add(key, Term(key=key, kind="pending", slug=slug, title=title, owner=owner))

    # --- Part 2: strong-candidate terms with no defining RFC. ---
    if include_candidates:
        for tag in _tag_universe(conn):
            low = tag.lower()
            # Conservative: multi-word tags only (same guard as titles).
            if " " not in tag and "\t" not in tag:
                continue
            if low in defined_titles or low in defined_slugs or _slugify(tag) in defined_slugs:
                continue
            add(low, Term(key=low, kind="candidate", term=tag))

    return LinkIndex(terms)