bot-bottle/bot_bottle/contrib/forge/base.py

"""Forge abstraction (PRD forge-native-integration, chunk 3).

The `Forge` abstract class is the provider-agnostic surface a forge
sidecar dispatches to: read issues/comments, post comments, edit
descriptions, and the membership / PR lookups the orchestrator needs.
Each forge (Gitea first) implements it; the sidecar protocol and the
agent prompt stay forge-agnostic.

`signal_done` is deliberately *not* a `Forge` method — completion is a
sidecar concept relayed to the orchestrator over a queue dir, not a
forge API operation.

`ScopedForge` enforces the PRD's **read-anywhere / write-scoped** model:
reads pass through to any issue/PR for context; writes are rejected
unless the target is the assigned issue or one of its PRs. This bounds
the blast radius of a prompt-injected agent below repo-wide API-key
permissions.
"""

from __future__ import annotations

import abc
from collections.abc import Iterable
from dataclasses import dataclass


@dataclass(frozen=True)
class Issue:
    """A forge issue or PR (forges model PRs as issues with the same
    number)."""

    number: int
    title: str
    body: str
    state: str  # "open" | "closed"


@dataclass(frozen=True)
class Comment:
    id: int
    user: str  # login of the comment author
    body: str


class ForgeScopeError(PermissionError):
    """Raised by `ScopedForge` when a write targets an issue/PR outside
    the assigned scope."""


class Forge(abc.ABC):
    """Provider-agnostic forge operations. Implementations wrap a
    per-provider HTTP client and translate to `Issue` / `Comment`."""

    @abc.abstractmethod
    def read_issue(self, number: int) -> Issue:
        """Read an issue or PR body (read-anywhere)."""

    @abc.abstractmethod
    def read_comments(self, number: int) -> list[Comment]:
        """Read a thread's comments (read-anywhere)."""

    @abc.abstractmethod
    def post_comment(self, number: int, body: str) -> None:
        """Post a comment to an issue or PR (write-scoped)."""

    @abc.abstractmethod
    def update_description(self, number: int, body: str) -> None:
        """Replace an issue or PR body (write-scoped)."""

    @abc.abstractmethod
    def is_org_member(self, org: str, username: str) -> bool:
        """Whether `username` is a member of `org`."""

    @abc.abstractmethod
    def get_pr_for_issue(self, number: int) -> int | None:
        """The PR number linked to an issue, or None when there is none."""

    @abc.abstractmethod
    def is_pr_open(self, number: int) -> bool:
        """Whether the given PR is still open."""


class ScopedForge(Forge):
    """Read-anywhere / write-scoped wrapper around a concrete `Forge`.

    `post_comment` and `update_description` are rejected with
    `ForgeScopeError` unless the target number is the assigned issue or
    one of the assigned PRs. Every other method delegates unchanged, so
    reads, membership checks, and PR lookups work against any number for
    context.

    The writable set is fixed at construction. The sidecar reconstructs
    a `ScopedForge` when a PR is discovered (`get_pr_for_issue`) so the
    new PR becomes writable; this class does not mutate its own scope.
    """

    def __init__(
        self,
        inner: Forge,
        *,
        assigned_issue: int,
        assigned_prs: Iterable[int] = (),
    ) -> None:
        self._inner = inner
        self._assigned_issue = assigned_issue
        self._writable = {assigned_issue, *assigned_prs}

    @property
    def writable(self) -> frozenset[int]:
        return frozenset(self._writable)

    def _check_write(self, number: int) -> None:
        if number not in self._writable:
            allowed = ", ".join(str(n) for n in sorted(self._writable))
            raise ForgeScopeError(
                f"write to #{number} denied: out of assigned scope "
                f"(writable: {allowed})"
            )

    # --- read-anywhere: pass through --------------------------------------

    def read_issue(self, number: int) -> Issue:
        return self._inner.read_issue(number)

    def read_comments(self, number: int) -> list[Comment]:
        return self._inner.read_comments(number)

    def is_org_member(self, org: str, username: str) -> bool:
        return self._inner.is_org_member(org, username)

    def get_pr_for_issue(self, number: int) -> int | None:
        return self._inner.get_pr_for_issue(number)

    def is_pr_open(self, number: int) -> bool:
        return self._inner.is_pr_open(number)

    # --- write-scoped: check then delegate --------------------------------

    def post_comment(self, number: int, body: str) -> None:
        self._check_write(number)
        self._inner.post_comment(number, body)

    def update_description(self, number: int, body: str) -> None:
        self._check_write(number)
        self._inner.update_description(number, body)