Forge (GitHub/GitLab)

Forge ABC

forge

Git forge abstraction for GitLab and GitHub.

Provides a polymorphic interface for merge/pull request operations, pipeline status checking, and review comment handling. Follows the same ABC pattern as agentic_ci.backend and agentic_ci.harness.

Usage::

from agentic_ci.forge import Forge

forge = Forge.detect("https://gitlab.com/org/repo/-/merge_requests/42")
status = forge.mr_status("https://gitlab.com/org/repo/-/merge_requests/42")

Label helpers (callers choose names and create-vs-attach policy)::

forge = Forge.detect(repo_url)
if not forge.label_exists(repo_url, "autofix"):
    forge.create_label(repo_url, "autofix")
forge.add_mr_labels(mr_url, ["autofix"])

Comment author trust (keep feedback only from repository owners, organization members and collaborators on GitHub, or Developers and above on GitLab; the GitHub check uses author_association, not repository permissions)::

forge = Forge.detect(mr_url, github_token=token)
threads = filter_trusted_threads(forge.review_comments(mr_url))
comments = filter_trusted_comments(forge.general_comments(mr_url))

GitHub comments carry author_association and GitLab comments carry author_access_level. Only TRUSTED_GITHUB_ASSOCIATIONS and levels at or above MIN_TRUSTED_GITLAB_ACCESS_LEVEL (Developer) are trusted. Accounts that do not meet that bar but whose feedback is wanted, such as review bots, are passed as trusted_authors and matched by account name (see :func:comment_author_id)::

threads = filter_trusted_threads(
    forge.review_comments(mr_url), trusted_authors={"coderabbitai[bot]"}
)

ForgeError

Bases: Exception

Raised when a forge API operation fails.

MergeRequestResult

Bases: tuple[str | None, str | None]

Result of creating or reusing a merge/pull request.

The result is a two-item (url, error) tuple, preserving the complete historical tuple contract. Consumers that need to distinguish a newly created request from a reused one should inspect :attr:created.

url property

Return the merge/pull request URL.

error property

Return the forge error, if any.

__new__(url, error, created=False)

Build a tuple-compatible result with creation metadata.

Source code in src/agentic_ci/forge/__init__.py

def __new__(
    cls,
    url: str | None,
    error: str | None,
    created: bool = False,
) -> MergeRequestResult:
    """Build a tuple-compatible result with creation metadata."""
    result = super().__new__(cls, (url, error))
    result.created = created
    return result

__getnewargs__()

Preserve created when copying or serializing the result.

Source code in src/agentic_ci/forge/__init__.py

def __getnewargs__(self) -> tuple[str | None, str | None, bool]:
    """Preserve ``created`` when copying or serializing the result."""
    return self.url, self.error, self.created

Forge

Bases: ABC

Abstract base for git forge (GitLab/GitHub) API operations.

Concrete implementations handle authentication and API differences. Use Forge.detect(url) to get the right implementation for a URL.

detect(url, *, github_token=None) classmethod

Return the correct Forge implementation for a URL.

Inspects the hostname to choose between GitLab and GitHub.

Parameters:

Name	Type	Description	Default
url	str	Any URL on the forge (repo URL, MR/PR URL, etc.).	required
github_token	str | None	Token for GitHub API authentication.	None

Raises:

Type	Description
ForgeError	If the URL hostname is not recognized.

Source code in src/agentic_ci/forge/__init__.py

@classmethod
def detect(cls, url: str, *, github_token: str | None = None) -> Forge:
    """Return the correct ``Forge`` implementation for a URL.

    Inspects the hostname to choose between GitLab and GitHub.

    Args:
        url: Any URL on the forge (repo URL, MR/PR URL, etc.).
        github_token: Token for GitHub API authentication.

    Raises:
        ForgeError: If the URL hostname is not recognized.
    """
    parsed = urlparse(url)
    if parsed.hostname == "gitlab.com":
        from agentic_ci.forge.gitlab import GitLabForge

        return GitLabForge()
    if parsed.hostname == "github.com":
        from agentic_ci.forge.github import GitHubForge

        return GitHubForge(token=github_token)
    raise ForgeError(f"Unrecognized forge host: {parsed.hostname} (URL: {url})")

create_merge_request(repo_url, source_branch, target_branch, title, description, draft=False) abstractmethod

Create an MR/PR.

Returns a :class:MergeRequestResult. The result's created field is true only when this call created the MR/PR; it is false when an existing open MR/PR was reused or the operation failed. The result continues to support url, error = unpacking for existing callers. Only an open MR/PR whose source branch lives in the target repository itself is reused, never one opened from a fork that happens to use the same branch name.

Source code in src/agentic_ci/forge/__init__.py

@abstractmethod
def create_merge_request(
    self,
    repo_url: str,
    source_branch: str,
    target_branch: str,
    title: str,
    description: str,
    draft: bool = False,
) -> MergeRequestResult:
    """Create an MR/PR.

    Returns a :class:`MergeRequestResult`.  The result's ``created`` field
    is true only when this call created the MR/PR; it is false when an
    existing open MR/PR was reused or the operation failed.  The result
    continues to support ``url, error =`` unpacking for existing callers.
    Only an open MR/PR whose source branch lives in the target repository
    itself is reused, never one opened from a fork that happens to use
    the same branch name.
    """

mr_status(mr_url, *, ignored_checks=None) abstractmethod

Get MR/PR state, source branch, repositories, and pipeline status.

Returns a dict with:

• state: "open", "merged" or "closed". GitLab's transient "locked" (an MR in the middle of being merged) is passed through, not reported as open.
• source_branch: the source branch name.
• pipeline_status: the overall CI status.
• source_repo / target_repo: opaque identities of the repository the source branch lives in and the repository the MR/PR targets (GitLab project IDs as strings, GitHub owner/name). They are equal, compared case-insensitively, exactly when the MR/PR is not from a fork; source_repo is "" when unknown (a deleted fork).
• target_repo_path: the target repository's current path (group/sub/repo or owner/name), which follows renames and transfers. It has the forge's canonical casing; compare it with a configured repository case-insensitively, as both forges treat paths.

When ignored_checks is provided, check runs whose name (or commit statuses whose context) is in the set are excluded before determining the overall pipeline status.

Source code in src/agentic_ci/forge/__init__.py

@abstractmethod
def mr_status(self, mr_url: str, *, ignored_checks: frozenset[str] | None = None) -> dict:
    """Get MR/PR state, source branch, repositories, and pipeline status.

    Returns a dict with:

    - ``state``: ``"open"``, ``"merged"`` or ``"closed"``. GitLab's
      transient ``"locked"`` (an MR in the middle of being merged) is
      passed through, not reported as open.
    - ``source_branch``: the source branch name.
    - ``pipeline_status``: the overall CI status.
    - ``source_repo`` / ``target_repo``: opaque identities of the
      repository the source branch lives in and the repository the
      MR/PR targets (GitLab project IDs as strings, GitHub
      ``owner/name``). They are equal, compared case-insensitively,
      exactly when the MR/PR is not from a fork; ``source_repo`` is
      ``""`` when unknown (a deleted fork).
    - ``target_repo_path``: the target repository's current path
      (``group/sub/repo`` or ``owner/name``), which follows renames and
      transfers. It has the forge's canonical casing; compare it with a
      configured repository case-insensitively, as both forges treat
      paths.

    When *ignored_checks* is provided, check runs whose ``name`` (or
    commit statuses whose ``context``) is in the set are excluded
    before determining the overall pipeline status.
    """

review_comments(mr_url) abstractmethod

Get unresolved review comment threads with diff positions.

Returns a list of dicts with keys: thread_id, file, line, body, author, comments.

body joins every comment as "<author>: <body>" lines and author is the thread starter. comments lists each comment of the thread as a dict with author, body, created_at and the author's trust field: author_association on GitHub, or author_username and author_access_level on GitLab. The thread dict also carries the starter's trust field. Use :func:filter_trusted_threads to drop untrusted comments.

Source code in src/agentic_ci/forge/__init__.py

@abstractmethod
def review_comments(self, mr_url: str) -> list[dict]:
    """Get unresolved review comment threads with diff positions.

    Returns a list of dicts with keys:
    ``thread_id``, ``file``, ``line``, ``body``, ``author``, ``comments``.

    ``body`` joins every comment as ``"<author>: <body>"`` lines and
    ``author`` is the thread starter. ``comments`` lists each comment
    of the thread as a dict with ``author``, ``body``, ``created_at``
    and the author's trust field: ``author_association`` on GitHub, or
    ``author_username`` and ``author_access_level`` on GitLab. The
    thread dict also carries the starter's trust field. Use
    :func:`filter_trusted_threads` to drop untrusted comments.
    """

general_comments(mr_url, since=None, skip_patterns=None) abstractmethod

Get general (non-diff-positioned) MR/PR comments.

Returns a list of dicts with keys: author, body, created_at, plus the author's trust field: author_association on GitHub, or author_username and author_access_level on GitLab. Use :func:filter_trusted_comments to drop untrusted comments. Comments created before since (ISO 8601) are excluded. Comments containing any string in skip_patterns are excluded. If skip_patterns is None, a default list is used.

Source code in src/agentic_ci/forge/__init__.py

@abstractmethod
def general_comments(
    self,
    mr_url: str,
    since: str | None = None,
    skip_patterns: list[str] | None = None,
) -> list[dict]:
    """Get general (non-diff-positioned) MR/PR comments.

    Returns a list of dicts with keys: ``author``, ``body``, ``created_at``,
    plus the author's trust field: ``author_association`` on GitHub, or
    ``author_username`` and ``author_access_level`` on GitLab. Use
    :func:`filter_trusted_comments` to drop untrusted comments.
    Comments created before ``since`` (ISO 8601) are excluded.
    Comments containing any string in ``skip_patterns`` are excluded.
    If ``skip_patterns`` is None, a default list is used.
    """

reply(mr_url, thread_id, message) abstractmethod

Reply to a review comment thread.

Source code in src/agentic_ci/forge/__init__.py

@abstractmethod
def reply(self, mr_url: str, thread_id: str, message: str) -> None:
    """Reply to a review comment thread."""

resolve(mr_url, thread_id) abstractmethod

Resolve a review comment thread.

Source code in src/agentic_ci/forge/__init__.py

@abstractmethod
def resolve(self, mr_url: str, thread_id: str) -> None:
    """Resolve a review comment thread."""

update_description(mr_url, *, title=None, description=None) abstractmethod

Update an existing MR/PR title and/or description.

Only the provided keyword arguments are updated; omitted fields are left unchanged.

Raises ForgeError on API failure.

Source code in src/agentic_ci/forge/__init__.py

@abstractmethod
def update_description(
    self,
    mr_url: str,
    *,
    title: str | None = None,
    description: str | None = None,
) -> None:
    """Update an existing MR/PR title and/or description.

    Only the provided keyword arguments are updated; omitted fields
    are left unchanged.

    Raises ``ForgeError`` on API failure.
    """

label_exists(repo_url, label) abstractmethod

Return whether label exists on the repository/project.

Does not create the label. Returns False when the label is missing (including HTTP 404). Raises ForgeError on other API failures.

Source code in src/agentic_ci/forge/__init__.py

@abstractmethod
def label_exists(self, repo_url: str, label: str) -> bool:
    """Return whether ``label`` exists on the repository/project.

    Does not create the label. Returns ``False`` when the label is
    missing (including HTTP 404). Raises ``ForgeError`` on other
    API failures.
    """

create_label(repo_url, label, *, color=None, description=None) abstractmethod

Create a repository/project label.

Callers must opt in explicitly; attaching labels does not implicitly create them via this API. When color is omitted, DEFAULT_LABEL_COLOR is used.

Raises ForgeError on API failure (including if the label already exists).

Source code in src/agentic_ci/forge/__init__.py

@abstractmethod
def create_label(
    self,
    repo_url: str,
    label: str,
    *,
    color: str | None = None,
    description: str | None = None,
) -> None:
    """Create a repository/project label.

    Callers must opt in explicitly; attaching labels does not
    implicitly create them via this API. When ``color`` is omitted,
    ``DEFAULT_LABEL_COLOR`` is used.

    Raises ``ForgeError`` on API failure (including if the label
    already exists).
    """

add_mr_labels(mr_url, labels) abstractmethod

Attach labels to an existing MR/PR, preserving current labels.

No-op when labels is empty. Does not check whether labels exist on the repository first — callers that need that policy should use :meth:label_exists / :meth:create_label.

Raises ForgeError on API failure.

Source code in src/agentic_ci/forge/__init__.py

@abstractmethod
def add_mr_labels(self, mr_url: str, labels: list[str]) -> None:
    """Attach labels to an existing MR/PR, preserving current labels.

    No-op when ``labels`` is empty. Does not check whether labels
    exist on the repository first — callers that need that policy
    should use :meth:`label_exists` / :meth:`create_label`.

    Raises ``ForgeError`` on API failure.
    """

pipeline_failures(mr_url, *, ignored_checks=None) abstractmethod

Get failed CI job names and log tails.

Returns {"pipeline_status": str, "failed_jobs": [{"name", "id", "log"}]}.

When ignored_checks is provided, those checks are excluded from both the pipeline status derivation and the failed_jobs list. An additional ignored_checks_status key reports the aggregate status of the ignored checks alone.

Source code in src/agentic_ci/forge/__init__.py

@abstractmethod
def pipeline_failures(
    self, mr_url: str, *, ignored_checks: frozenset[str] | None = None
) -> dict:
    """Get failed CI job names and log tails.

    Returns ``{"pipeline_status": str, "failed_jobs": [{"name", "id", "log"}]}``.

    When *ignored_checks* is provided, those checks are excluded from
    both the pipeline status derivation and the ``failed_jobs`` list.
    An additional ``ignored_checks_status`` key reports the aggregate
    status of the ignored checks alone.
    """

comment_author_id(comment)

Return the account name that trusted_authors matches, lowercased.

A GitHub comment (one with an author_association key) is identified by its login in author, which ends in [bot] for a GitHub App, so an App cannot be confused with a user account of the same name. A GitLab comment is identified by author_username, never by author, which is the display name any user can set. Returns "" when the account is unknown (a deleted account, for example).

Source code in src/agentic_ci/forge/__init__.py

def comment_author_id(comment: dict) -> str:
    """Return the account name that ``trusted_authors`` matches, lowercased.

    A GitHub comment (one with an ``author_association`` key) is identified
    by its login in ``author``, which ends in ``[bot]`` for a GitHub App, so
    an App cannot be confused with a user account of the same name. A
    GitLab comment is identified by ``author_username``, never by
    ``author``, which is the display name any user can set. Returns ``""``
    when the account is unknown (a deleted account, for example).
    """
    if "author_association" in comment:
        login = comment.get("author") or ""
        return "" if login == "Unknown" else login.lower()
    return (comment.get("author_username") or "").lower()

is_trusted_comment(comment, trusted_authors=())

Return whether a forge comment was written by a trusted author.

A comment whose :func:comment_author_id is in trusted_authors (account names, compared without case) is trusted. Otherwise a GitHub comment (one with an author_association key) is trusted when the association is in TRUSTED_GITHUB_ASSOCIATIONS, and any other comment only when its author_access_level is an integer at or above MIN_TRUSTED_GITLAB_ACCESS_LEVEL. Comments carrying neither field are untrusted, so the check fails closed.

Source code in src/agentic_ci/forge/__init__.py

def is_trusted_comment(comment: dict, trusted_authors: Iterable[str] = ()) -> bool:
    """Return whether a forge comment was written by a trusted author.

    A comment whose :func:`comment_author_id` is in *trusted_authors*
    (account names, compared without case) is trusted. Otherwise a GitHub
    comment (one with an ``author_association`` key) is trusted when the
    association is in ``TRUSTED_GITHUB_ASSOCIATIONS``, and any other comment
    only when its ``author_access_level`` is an integer at or above
    ``MIN_TRUSTED_GITLAB_ACCESS_LEVEL``. Comments carrying neither field
    are untrusted, so the check fails closed.
    """
    author = comment_author_id(comment)
    if author and author in _normalized_authors(trusted_authors):
        return True
    if "author_association" in comment:
        return comment["author_association"] in TRUSTED_GITHUB_ASSOCIATIONS
    level = comment.get("author_access_level")
    if isinstance(level, bool) or not isinstance(level, int):
        return False
    return level >= MIN_TRUSTED_GITLAB_ACCESS_LEVEL

filter_trusted_comments(comments, trusted_authors=())

Return only the comments written by trusted authors.

Takes the output of :meth:Forge.general_comments (or any list of comment dicts) and keeps those for which :func:is_trusted_comment holds, with trusted_authors as extra trusted account names.

Source code in src/agentic_ci/forge/__init__.py

def filter_trusted_comments(
    comments: list[dict], trusted_authors: Iterable[str] = ()
) -> list[dict]:
    """Return only the comments written by trusted authors.

    Takes the output of :meth:`Forge.general_comments` (or any list of
    comment dicts) and keeps those for which :func:`is_trusted_comment`
    holds, with *trusted_authors* as extra trusted account names.
    """
    authors = _normalized_authors(trusted_authors)
    return [c for c in comments if is_trusted_comment(c, authors)]

filter_trusted_threads(threads, trusted_authors=())

Return review threads reduced to their trusted comments.

Takes the output of :meth:Forge.review_comments. Each thread's comments keeps only trusted comments and body is rebuilt from them, so an untrusted reply inside a trusted author's thread is dropped. A thread with no trusted comment left, or without a comments list, is dropped. The thread's author fields (author, author_username, author_access_level, author_association) are taken from the first trusted comment, so a thread an outsider started carries none of the outsider's account data (a GitLab display name is free text). Of the other thread fields only thread_id, file and line are kept; any other key is dropped. trusted_authors are extra trusted account names (see :func:is_trusted_comment). Input dicts are not modified.

Source code in src/agentic_ci/forge/__init__.py

def filter_trusted_threads(threads: list[dict], trusted_authors: Iterable[str] = ()) -> list[dict]:
    """Return review threads reduced to their trusted comments.

    Takes the output of :meth:`Forge.review_comments`. Each thread's
    ``comments`` keeps only trusted comments and ``body`` is rebuilt from
    them, so an untrusted reply inside a trusted author's thread is
    dropped. A thread with no trusted comment left, or without a
    ``comments`` list, is dropped. The thread's author fields (``author``,
    ``author_username``, ``author_access_level``, ``author_association``)
    are taken from the first trusted comment, so a thread an outsider
    started carries none of the outsider's account data (a GitLab display
    name is free text). Of the other thread fields only ``thread_id``,
    ``file`` and ``line`` are kept; any other key is dropped.
    *trusted_authors* are extra trusted account names (see
    :func:`is_trusted_comment`). Input dicts are not modified.
    """
    authors = _normalized_authors(trusted_authors)
    result: list[dict] = []
    for thread in threads:
        trusted = filter_trusted_comments(thread.get("comments") or [], authors)
        if not trusted:
            continue
        body = "\n".join(f"{c.get('author', 'Unknown')}: {c.get('body', '')}" for c in trusted)
        position = {k: thread[k] for k in _THREAD_POSITION_KEYS if k in thread}
        author = {k: trusted[0][k] for k in _THREAD_AUTHOR_KEYS if k in trusted[0]}
        result.append({**position, **author, "comments": trusted, "body": body})
    return result

parse_gitlab_mr_url(url)

Parse a GitLab MR URL into (project_path, mr_iid).

Raises ForgeError if the URL does not match the expected pattern.

Source code in src/agentic_ci/forge/__init__.py

def parse_gitlab_mr_url(url: str) -> tuple[str, int]:
    """Parse a GitLab MR URL into ``(project_path, mr_iid)``.

    Raises ``ForgeError`` if the URL does not match the expected pattern.
    """
    match = _GITLAB_MR_RE.match(url)
    if not match:
        raise ForgeError(f"Invalid GitLab MR URL: {url}")
    return match.group(1), int(match.group(2))

parse_github_pr_url(url)

Parse a GitHub PR URL into (owner/repo, pr_number).

Raises ForgeError if the URL does not match the expected pattern.

Source code in src/agentic_ci/forge/__init__.py

def parse_github_pr_url(url: str) -> tuple[str, int]:
    """Parse a GitHub PR URL into ``(owner/repo, pr_number)``.

    Raises ``ForgeError`` if the URL does not match the expected pattern.
    """
    match = _GITHUB_PR_RE.match(url)
    if not match:
        raise ForgeError(f"Invalid GitHub PR URL: {url}")
    return match.group(1), int(match.group(2))

repo_path_from_url(url)

Extract the repository path from a GitLab or GitHub URL.

Strips trailing slashes and .git suffixes.

Source code in src/agentic_ci/forge/__init__.py

def repo_path_from_url(url: str) -> str:
    """Extract the repository path from a GitLab or GitHub URL.

    Strips trailing slashes and ``.git`` suffixes.
    """
    parsed = urlparse(url)
    path = parsed.path.strip("/")
    if path.endswith(".git"):
        path = path[:-4]
    return path

detect_forge(url, *, github_token=None)

Convenience wrapper around Forge.detect().

Source code in src/agentic_ci/forge/__init__.py

def detect_forge(url: str, *, github_token: str | None = None) -> Forge:
    """Convenience wrapper around ``Forge.detect()``."""
    return Forge.detect(url, github_token=github_token)

GitHub

github

GitHub forge implementation.

Provides GitHubForge for interacting with the GitHub REST and GraphQL APIs (pull requests, check runs, review threads).

GitHubForge(token=None)

Bases: Forge

GitHub REST + GraphQL API implementation of the Forge interface.

Source code in src/agentic_ci/forge/github.py

def __init__(self, token: str | None = None) -> None:
    self._token = token
    self._session = build_session(github_token=token)

graphql(query, variables=None)

Execute a GitHub GraphQL query or mutation.

Returns the data dict from the response. Raises ForgeError on HTTP or GraphQL errors.

Source code in src/agentic_ci/forge/github.py

def graphql(self, query: str, variables: dict | None = None) -> dict:
    """Execute a GitHub GraphQL query or mutation.

    Returns the ``data`` dict from the response.
    Raises ``ForgeError`` on HTTP or GraphQL errors.
    """
    payload: dict = {"query": query}
    if variables:
        payload["variables"] = variables
    resp = self._session.post("https://api.github.com/graphql", json=payload)
    if resp.status_code != 200:
        raise ForgeError(f"GraphQL HTTP {resp.status_code}: {resp.text}")
    data = resp.json()
    if "errors" in data:
        raise ForgeError(f"GraphQL errors: {json.dumps(data['errors'])}")
    return data["data"]

check_runs(repo_path, sha)

Fetch all check runs for a commit SHA.

Returns (check_runs, accessible) where accessible is False when the token lacks checks:read permission (403).

Source code in src/agentic_ci/forge/github.py

def check_runs(self, repo_path: str, sha: str) -> tuple[list[dict], bool]:
    """Fetch all check runs for a commit SHA.

    Returns ``(check_runs, accessible)`` where ``accessible`` is
    False when the token lacks ``checks:read`` permission (403).
    """
    all_runs: list[dict] = []
    page = 1
    while True:
        resp = self._session.get(
            f"https://api.github.com/repos/{repo_path}/commits/{sha}/check-runs",
            params={"per_page": 100, "page": page},
        )
        if resp.status_code == 403:
            log.warning("checks:read permission not available: %s", resp.text)
            return [], False
        if resp.status_code != 200:
            raise ForgeError(f"HTTP {resp.status_code} fetching check runs: {resp.text}")
        runs = resp.json().get("check_runs", [])
        all_runs.extend(runs)
        if len(runs) < 100:
            break
        page += 1
    return all_runs, True

commit_statuses(repo_path, sha)

Fetch commit statuses for a SHA.

GitHub has two parallel CI reporting mechanisms: Check Runs (used by GitHub Actions) and Commit Statuses (the older API used by external CI like Prow, Jenkins, and other integrations). check_runs() only covers the first; this method covers the second so _derive_pipeline_status() sees the full picture.

Merge-management contexts (tide, Mergify) are excluded since they reflect merge policy, not CI results.

See https://docs.github.com/en/rest/commits/statuses

Source code in src/agentic_ci/forge/github.py

def commit_statuses(self, repo_path: str, sha: str) -> list[dict]:
    """Fetch commit statuses for a SHA.

    GitHub has two parallel CI reporting mechanisms: Check Runs
    (used by GitHub Actions) and Commit Statuses (the older API
    used by external CI like Prow, Jenkins, and other integrations).
    ``check_runs()`` only covers the first; this method covers the
    second so ``_derive_pipeline_status()`` sees the full picture.

    Merge-management contexts (``tide``, ``Mergify``) are excluded
    since they reflect merge policy, not CI results.

    See https://docs.github.com/en/rest/commits/statuses
    """
    all_statuses: list[dict] = []
    page = 1
    while True:
        resp = self._session.get(
            f"https://api.github.com/repos/{repo_path}/commits/{sha}/statuses",
            params={"per_page": 100, "page": page},
        )
        if resp.status_code == 403:
            log.debug("statuses not accessible for %s: %s", sha, resp.text)
            return []
        if resp.status_code != 200:
            log.warning("HTTP %d fetching commit statuses: %s", resp.status_code, resp.text)
            return []
        batch = resp.json()
        if not batch:
            break
        all_statuses.extend(batch)
        if len(batch) < 100:
            break
        page += 1
    seen: dict[str, dict] = {}
    for s in all_statuses:
        ctx = s.get("context", "")
        if ctx not in seen:
            seen[ctx] = s
    return [s for s in seen.values() if not _is_merge_management_status(s.get("context", ""))]

generate_github_jwt(app_id, private_key_pem)

Generate a GitHub App JWT signed with RS256.

The JWT is valid for 10 minutes (GitHub maximum).

Requires the PyJWT and cryptography packages. Install with: pip install agentic-ci[forge]

Source code in src/agentic_ci/forge/github.py

def generate_github_jwt(app_id: str | int, private_key_pem: str) -> str:
    """Generate a GitHub App JWT signed with RS256.

    The JWT is valid for 10 minutes (GitHub maximum).

    Requires the ``PyJWT`` and ``cryptography`` packages.
    Install with: ``pip install agentic-ci[forge]``
    """
    import jwt  # type: ignore[import-not-found]

    now = int(time.time())
    payload = {
        "iat": now - 60,
        "exp": now + (10 * 60),
        "iss": str(app_id),
    }
    return jwt.encode(payload, private_key_pem, algorithm="RS256")

get_installation_token(jwt_token, installation_id)

Exchange a GitHub App JWT for an installation access token.

Returns the token string (valid for 1 hour). Raises RuntimeError on failure.

Source code in src/agentic_ci/forge/github.py

def get_installation_token(jwt_token: str, installation_id: str | int) -> str:
    """Exchange a GitHub App JWT for an installation access token.

    Returns the token string (valid for 1 hour).
    Raises ``RuntimeError`` on failure.
    """
    resp = requests.post(
        f"https://api.github.com/app/installations/{installation_id}/access_tokens",
        headers={
            "Authorization": f"Bearer {jwt_token}",
            "Accept": "application/vnd.github+json",
        },
        timeout=API_TIMEOUT,
    )
    if resp.status_code != 201:
        raise RuntimeError(f"HTTP {resp.status_code} creating installation token: {resp.text}")
    return resp.json()["token"]

resolve_app_token(repo_url, github_config)

Resolve a GitHub App installation token for a repo URL.

Extracts the GitHub org from repo_url, looks up the matching App configuration in github_config, and returns a short-lived installation token.

Returns None (with an error log) on any failure.

Source code in src/agentic_ci/forge/github.py

def resolve_app_token(repo_url: str, github_config: dict) -> str | None:
    """Resolve a GitHub App installation token for a repo URL.

    Extracts the GitHub org from ``repo_url``, looks up the matching
    App configuration in ``github_config``, and returns a short-lived
    installation token.

    Returns ``None`` (with an error log) on any failure.
    """
    match = _GITHUB_ORG_RE.search(repo_url)
    if not match:
        log.error("Cannot extract GitHub org from URL: %s", repo_url)
        return None
    org_name = match.group(1).lower()
    app_config = None
    for key, value in github_config.items():
        if key.lower() == org_name:
            app_config = value
            break
    if not app_config:
        log.error("No GitHub App configured for org '%s'", org_name)
        return None
    credentials_env = app_config.get("credentials_env", "")
    private_key_file = app_config.get("private_key_file", "")
    if not credentials_env or not private_key_file:
        log.error("Incomplete GitHub App config for org '%s'", org_name)
        return None
    credentials_json = os.environ.get(credentials_env, "")
    if not credentials_json:
        log.error("GitHub App env var %s is not set", credentials_env)
        return None
    try:
        creds = json.loads(credentials_json)
    except (json.JSONDecodeError, TypeError):
        log.error("GitHub App env var %s is not valid JSON", credentials_env)
        return None
    app_id = creds.get("app_id", "")
    installation_id = creds.get("installation_id", "")
    if not app_id or not installation_id:
        log.error("GitHub App credentials missing app_id or installation_id")
        return None
    key_path = _find_private_key(private_key_file)
    if not key_path:
        log.error("GitHub App private key file not found: %s", private_key_file)
        return None
    try:
        private_key_pem = key_path.read_text(encoding="utf-8")
    except OSError as exc:
        log.error("Cannot read private key %s: %s", key_path, exc)
        return None
    try:
        jwt_token = generate_github_jwt(app_id, private_key_pem)
        return get_installation_token(jwt_token, installation_id)
    except Exception as exc:
        log.error("GitHub token generation failed: %s", exc)
        return None

GitLab

gitlab

GitLab forge implementation.

Provides GitLabForge for interacting with the GitLab REST API (merge requests, pipelines, discussions).

GitLabForge(*, adapter=None)

Bases: Forge

GitLab REST API implementation of the Forge interface.

Source code in src/agentic_ci/forge/gitlab.py

def __init__(self, *, adapter: GitLabHTTPAdapter | None = None) -> None:
    self._session = build_session(gitlab_adapter=adapter)
    # (project_id, user_id) -> access level, so each distinct comment
    # author costs at most one members API call per client.
    self._access_levels: dict[tuple[int, int], int | None] = {}

session property

The authenticated HTTP session for direct API calls.

project_id(project_path)

Look up the numeric GitLab project ID from a project path.

Raises ForgeError if the project cannot be found.

Source code in src/agentic_ci/forge/gitlab.py

def project_id(self, project_path: str) -> int:
    """Look up the numeric GitLab project ID from a project path.

    Raises ``ForgeError`` if the project cannot be found.
    """
    encoded = urllib.parse.quote(project_path, safe="")
    resp = self._session.get(f"https://gitlab.com/api/v4/projects/{encoded}")
    if resp.status_code != 200:
        raise ForgeError(
            f"HTTP {resp.status_code} looking up project {project_path}: {resp.text}"
        )
    return resp.json()["id"]

member_access_level(pid, user_id)

Return a user's effective access level on a project.

Uses members/all so levels inherited from parent groups count. Returns 0 when the user is not a member or the membership is not active (for example awaiting seat approval or blocked), and None when the level could not be resolved (for example HTTP 403). Results are cached per client, so repeated calls cost one request per (pid, user_id).

Source code in src/agentic_ci/forge/gitlab.py

def member_access_level(self, pid: int, user_id: int) -> int | None:
    """Return a user's effective access level on a project.

    Uses ``members/all`` so levels inherited from parent groups count.
    Returns ``0`` when the user is not a member or the membership is not
    ``active`` (for example ``awaiting`` seat approval or ``blocked``),
    and ``None`` when the level could not be resolved (for example HTTP
    403). Results are cached per client, so repeated calls cost one
    request per ``(pid, user_id)``.
    """
    key = (pid, user_id)
    if key in self._access_levels:
        return self._access_levels[key]
    resp = self._session.get(
        f"https://gitlab.com/api/v4/projects/{pid}/members/all/{user_id}",
    )
    level: int | None
    if resp.status_code == 200:
        member = resp.json()
        raw = member.get("access_level")
        state = member.get("state")
        if state is not None and state != "active":
            level = 0
        elif isinstance(raw, int) and not isinstance(raw, bool):
            level = raw
        else:
            level = None
    elif resp.status_code == 404:
        level = 0
    else:
        log.warning(
            "HTTP %d resolving access level of user %s on project %s: %s",
            resp.status_code,
            user_id,
            pid,
            _sanitize_resp_text(resp.text),
        )
        level = None
    self._access_levels[key] = level
    return level

pipeline_metadata(project_path, pipeline_id)

Get pipeline metadata.

Returns the raw pipeline dict from the GitLab API.

Source code in src/agentic_ci/forge/gitlab.py

def pipeline_metadata(self, project_path: str, pipeline_id: int) -> dict:
    """Get pipeline metadata.

    Returns the raw pipeline dict from the GitLab API.
    """
    pipeline_id = int(pipeline_id)
    pid = self.project_id(project_path)
    resp = self._session.get(
        f"https://gitlab.com/api/v4/projects/{pid}/pipelines/{pipeline_id}",
    )
    if resp.status_code != 200:
        raise ForgeError(f"HTTP {resp.status_code}: {resp.text}")
    return resp.json()

pipeline_jobs(project_path, pipeline_id, *, scope=None)

Get all jobs for a pipeline (paginated).

Parameters:

Name	Type	Description	Default
project_path	str	GitLab project path (e.g. "org/repo").	required
pipeline_id	int	Numeric pipeline ID.	required
scope	str | None	Optional job scope filter (e.g. "failed").	None

Returns a list of raw job dicts from the GitLab API.

Source code in src/agentic_ci/forge/gitlab.py

def pipeline_jobs(
    self, project_path: str, pipeline_id: int, *, scope: str | None = None
) -> list[dict]:
    """Get all jobs for a pipeline (paginated).

    Args:
        project_path: GitLab project path (e.g. ``"org/repo"``).
        pipeline_id: Numeric pipeline ID.
        scope: Optional job scope filter (e.g. ``"failed"``).

    Returns a list of raw job dicts from the GitLab API.
    """
    pipeline_id = int(pipeline_id)
    pid = self.project_id(project_path)
    params: dict[str, str] = {}
    if scope:
        params["scope[]"] = scope
    return self._paginate(
        f"https://gitlab.com/api/v4/projects/{pid}/pipelines/{pipeline_id}/jobs",
        params=params,
    )

job_trace(project_path, job_id)

Get the full trace log for a job.

Parameters:

Name	Type	Description	Default
project_path	str	GitLab project path (e.g. "org/repo").	required
job_id	int	Numeric job ID.	required

Returns the trace text as a string.

Source code in src/agentic_ci/forge/gitlab.py

def job_trace(self, project_path: str, job_id: int) -> str:
    """Get the full trace log for a job.

    Args:
        project_path: GitLab project path (e.g. ``"org/repo"``).
        job_id: Numeric job ID.

    Returns the trace text as a string.
    """
    job_id = int(job_id)
    pid = self.project_id(project_path)
    resp = self._session.get(
        f"https://gitlab.com/api/v4/projects/{pid}/jobs/{job_id}/trace",
    )
    if resp.status_code != 200:
        raise ForgeError(f"HTTP {resp.status_code}: {resp.text}")
    return resp.text

retry_job(project_path, job_id)

Retry a failed or canceled GitLab CI job.

Parameters:

Name	Type	Description	Default
project_path	str	GitLab project path (e.g. "org/repo").	required
job_id	int	Numeric job ID to retry.	required

Returns the new job dict from the API response (includes the retried job's id, web_url, and status).

Raises:

Type	Description
ForgeError	On API failure. A 401/403 response raises an actionable error indicating the token needs the api or write_build scope to retry jobs.

Source code in src/agentic_ci/forge/gitlab.py

def retry_job(self, project_path: str, job_id: int) -> dict:
    """Retry a failed or canceled GitLab CI job.

    Args:
        project_path: GitLab project path (e.g. ``"org/repo"``).
        job_id: Numeric job ID to retry.

    Returns the new job dict from the API response (includes the
    retried job's ``id``, ``web_url``, and ``status``).

    Raises:
        ForgeError: On API failure. A 401/403 response raises an
            actionable error indicating the token needs the ``api``
            or ``write_build`` scope to retry jobs.
    """
    job_id = int(job_id)
    pid = self.project_id(project_path)
    resp = self._session.post(
        f"https://gitlab.com/api/v4/projects/{pid}/jobs/{job_id}/retry",
    )
    if resp.status_code in (401, 403):
        raise ForgeError(
            f"GitLab API returned {resp.status_code}: token does not have "
            "permission to retry jobs. Ensure BOT_PAT has the 'api' or "
            "'write_build' scope."
        )
    if resp.status_code not in (200, 201):
        raise ForgeError(f"HTTP {resp.status_code}: {_sanitize_resp_text(resp.text)}")
    return resp.json()

pipeline_schedules(project_path)

List all pipeline schedules for a project (paginated).

Parameters:

Name	Type	Description	Default
project_path	str	GitLab project path (e.g. "org/repo").	required

Returns a list of raw schedule dicts from the GitLab API.

Source code in src/agentic_ci/forge/gitlab.py

def pipeline_schedules(self, project_path: str) -> list[dict]:
    """List all pipeline schedules for a project (paginated).

    Args:
        project_path: GitLab project path (e.g. ``"org/repo"``).

    Returns a list of raw schedule dicts from the GitLab API.
    """
    pid = self.project_id(project_path)
    return self._paginate(
        f"https://gitlab.com/api/v4/projects/{pid}/pipeline_schedules",
    )

pipeline_schedule(project_path, schedule_id)

Get details of a specific pipeline schedule.

The returned dict includes last_pipeline with the most recent pipeline triggered by this schedule.

Parameters:

Name	Type	Description	Default
project_path	str	GitLab project path (e.g. "org/repo").	required
schedule_id	int	Numeric pipeline schedule ID.	required

Returns the raw schedule dict from the GitLab API.

Source code in src/agentic_ci/forge/gitlab.py

def pipeline_schedule(self, project_path: str, schedule_id: int) -> dict:
    """Get details of a specific pipeline schedule.

    The returned dict includes ``last_pipeline`` with the most recent
    pipeline triggered by this schedule.

    Args:
        project_path: GitLab project path (e.g. ``"org/repo"``).
        schedule_id: Numeric pipeline schedule ID.

    Returns the raw schedule dict from the GitLab API.
    """
    schedule_id = int(schedule_id)
    pid = self.project_id(project_path)
    resp = self._session.get(
        f"https://gitlab.com/api/v4/projects/{pid}/pipeline_schedules/{schedule_id}",
    )
    if resp.status_code != 200:
        raise ForgeError(f"HTTP {resp.status_code}: {resp.text}")
    return resp.json()

add_mr_block(blocked_mr_url, blocking_mr_url)

Set MR dependency: blocked_mr cannot merge until blocking_mr merges.

Requires GitLab Premium.

Parameters:

Name	Type	Description	Default
blocked_mr_url	str	URL of the MR that should be blocked.	required
blocking_mr_url	str	URL of the MR that must merge first.	required

Raises ForgeError on failure.

Source code in src/agentic_ci/forge/gitlab.py

def add_mr_block(self, blocked_mr_url: str, blocking_mr_url: str) -> None:
    """Set MR dependency: *blocked_mr* cannot merge until *blocking_mr* merges.

    Requires GitLab Premium.

    Args:
        blocked_mr_url: URL of the MR that should be blocked.
        blocking_mr_url: URL of the MR that must merge first.

    Raises ``ForgeError`` on failure.
    """
    blocked_path, blocked_iid = parse_gitlab_mr_url(blocked_mr_url)
    blocking_path, blocking_iid = parse_gitlab_mr_url(blocking_mr_url)

    blocking_pid = self.project_id(blocking_path)
    resp = self._session.get(
        f"https://gitlab.com/api/v4/projects/{blocking_pid}/merge_requests/{blocking_iid}",
    )
    if resp.status_code != 200:
        raise ForgeError(f"HTTP {resp.status_code} fetching blocking MR: {resp.text}")
    blocking_internal_id = resp.json()["id"]

    blocked_pid = self.project_id(blocked_path)
    resp = self._session.post(
        f"https://gitlab.com/api/v4/projects/{blocked_pid}/merge_requests/{blocked_iid}/blocks",
        json={"blocking_merge_request_id": blocking_internal_id},
    )
    if resp.status_code not in (200, 201):
        raise ForgeError(f"HTTP {resp.status_code} creating MR block: {resp.text}")

mr_diff_position(mr_url)

Get the first changed line position and diff refs from a GitLab MR.

This is a GitLab-specific operation not available on other forges. Returns {"file", "line", "base_sha", "head_sha", "start_sha"}.

Source code in src/agentic_ci/forge/gitlab.py

def mr_diff_position(self, mr_url: str) -> dict:
    """Get the first changed line position and diff refs from a GitLab MR.

    This is a GitLab-specific operation not available on other forges.
    Returns ``{"file", "line", "base_sha", "head_sha", "start_sha"}``.
    """
    project_path, mr_iid = parse_gitlab_mr_url(mr_url)
    pid = self.project_id(project_path)
    resp = self._session.get(
        f"https://gitlab.com/api/v4/projects/{pid}/merge_requests/{mr_iid}/changes",
    )
    if resp.status_code != 200:
        raise ForgeError(f"HTTP {resp.status_code}: {resp.text}")
    data = resp.json()
    diff_refs = data.get("diff_refs", {})
    result = {
        "file": "",
        "line": 0,
        "base_sha": diff_refs.get("base_sha", ""),
        "head_sha": diff_refs.get("head_sha", ""),
        "start_sha": diff_refs.get("start_sha", ""),
    }
    for change in data.get("changes", []):
        diff_text = change.get("diff", "")
        new_path = change.get("new_path", "")
        if not diff_text or not new_path:
            continue
        line = _find_first_added_line(diff_text)
        if line is not None:
            result["file"] = new_path
            result["line"] = line
            break
    return result

paginate(url, params=None)

Fetch all pages of a paginated GitLab API endpoint.

Source code in src/agentic_ci/forge/gitlab.py

def paginate(self, url: str, params: dict | None = None) -> list[dict]:
    """Fetch all pages of a paginated GitLab API endpoint."""
    all_items: list[dict] = []
    page = 1
    while True:
        page_params = {"per_page": 100, "page": page}
        if params:
            page_params.update(params)
        resp = self._session.get(url, params=page_params)
        if resp.status_code != 200:
            raise ForgeError(f"HTTP {resp.status_code}: {resp.text}")
        batch = resp.json()
        if not batch:
            break
        all_items.extend(batch)
        if len(batch) < 100:
            break
        page += 1
    return all_items

CLI

cli

CLI subcommands for forge operations.

Registered as the agentic-ci forge subcommand group.

Usage::

agentic-ci forge mr-status <URL>
agentic-ci forge mr-comments <URL>
agentic-ci forge mr-general-comments <URL> [--since ISO]
agentic-ci forge mr-reply <URL> <thread_id> <message>
agentic-ci forge mr-resolve <URL> <thread_id>
agentic-ci forge pipeline-failures <URL>
agentic-ci forge mr-diff-position <URL>
agentic-ci forge label-exists <REPO_URL> <label>
agentic-ci forge label-create <REPO_URL> <label> [--color HEX] [--description TEXT]
agentic-ci forge mr-add-labels <URL> --labels LABEL [LABEL ...]
agentic-ci forge github-token --app-id ID --installation-id ID --private-key PEM

cmd_mr_status(args)

Get MR/PR state, source branch, and pipeline status.

Source code in src/agentic_ci/forge/cli.py

def cmd_mr_status(args: argparse.Namespace) -> None:
    """Get MR/PR state, source branch, and pipeline status."""
    forge = Forge.detect(args.url, github_token=args.token)
    result = forge.mr_status(args.url)
    json.dump(result, sys.stdout, indent=2)
    print()

cmd_mr_comments(args)

Get unresolved review comment threads.

Source code in src/agentic_ci/forge/cli.py

def cmd_mr_comments(args: argparse.Namespace) -> None:
    """Get unresolved review comment threads."""
    forge = Forge.detect(args.url, github_token=args.token)
    result = forge.review_comments(args.url)
    json.dump(result, sys.stdout, indent=2)
    print()

cmd_mr_general_comments(args)

Get general (non-diff-positioned) MR/PR comments.

Source code in src/agentic_ci/forge/cli.py

def cmd_mr_general_comments(args: argparse.Namespace) -> None:
    """Get general (non-diff-positioned) MR/PR comments."""
    forge = Forge.detect(args.url, github_token=args.token)
    result = forge.general_comments(args.url, since=args.since)
    json.dump(result, sys.stdout, indent=2)
    print()

cmd_mr_reply(args)

Reply to a review thread.

Source code in src/agentic_ci/forge/cli.py

def cmd_mr_reply(args: argparse.Namespace) -> None:
    """Reply to a review thread."""
    forge = Forge.detect(args.url, github_token=args.token)
    forge.reply(args.url, args.thread_id, args.message)
    print(f"Replied to thread {args.thread_id}")

cmd_mr_resolve(args)

Resolve a review thread.

Source code in src/agentic_ci/forge/cli.py

def cmd_mr_resolve(args: argparse.Namespace) -> None:
    """Resolve a review thread."""
    forge = Forge.detect(args.url, github_token=args.token)
    forge.resolve(args.url, args.thread_id)
    print(f"Resolved thread {args.thread_id}")

cmd_pipeline_failures(args)

Get failed CI job names and logs.

Source code in src/agentic_ci/forge/cli.py

def cmd_pipeline_failures(args: argparse.Namespace) -> None:
    """Get failed CI job names and logs."""
    forge = Forge.detect(args.url, github_token=args.token)
    result = forge.pipeline_failures(args.url)
    json.dump(result, sys.stdout, indent=2)
    print()

cmd_mr_update(args)

Update MR/PR title and/or description.

Source code in src/agentic_ci/forge/cli.py

def cmd_mr_update(args: argparse.Namespace) -> None:
    """Update MR/PR title and/or description."""
    forge = Forge.detect(args.url, github_token=args.token)
    forge.update_description(args.url, title=args.title, description=args.description)
    print(f"Updated {args.url}")

cmd_label_exists(args)

Check whether a repository/project label exists.

Source code in src/agentic_ci/forge/cli.py

def cmd_label_exists(args: argparse.Namespace) -> None:
    """Check whether a repository/project label exists."""
    forge = Forge.detect(args.url, github_token=args.token)
    exists = forge.label_exists(args.url, args.label)
    json.dump(exists, sys.stdout)
    print()

cmd_label_create(args)

Create a repository/project label.

Source code in src/agentic_ci/forge/cli.py

def cmd_label_create(args: argparse.Namespace) -> None:
    """Create a repository/project label."""
    forge = Forge.detect(args.url, github_token=args.token)
    forge.create_label(
        args.url,
        args.label,
        color=args.color,
        description=args.description,
    )
    print(f"Created label {args.label!r} on {args.url}")

cmd_mr_add_labels(args)

Attach labels to an existing MR/PR.

Source code in src/agentic_ci/forge/cli.py

def cmd_mr_add_labels(args: argparse.Namespace) -> None:
    """Attach labels to an existing MR/PR."""
    forge = Forge.detect(args.url, github_token=args.token)
    forge.add_mr_labels(args.url, args.labels)
    print(f"Added labels {args.labels} to {args.url}")

cmd_mr_diff_position(args)

Get the first changed line position and diff refs (GitLab only).

Source code in src/agentic_ci/forge/cli.py

def cmd_mr_diff_position(args: argparse.Namespace) -> None:
    """Get the first changed line position and diff refs (GitLab only)."""
    forge = Forge.detect(args.url, github_token=args.token)
    if not isinstance(forge, GitLabForge):
        print("Error: mr-diff-position is only supported for GitLab", file=sys.stderr)
        sys.exit(1)
    result = forge.mr_diff_position(args.url)
    json.dump(result, sys.stdout, indent=2)
    print()

cmd_github_token(args)

Generate a short-lived GitHub App installation token.

Source code in src/agentic_ci/forge/cli.py

def cmd_github_token(args: argparse.Namespace) -> None:
    """Generate a short-lived GitHub App installation token."""
    private_key = args.private_key
    if os.path.isfile(private_key):
        try:
            with open(private_key, encoding="utf-8") as f:
                private_key = f.read()
        except OSError as exc:
            print(f"Error: failed to read private key file: {exc}", file=sys.stderr)
            sys.exit(1)
    jwt_token = generate_github_jwt(args.app_id, private_key)
    token = get_installation_token(jwt_token, args.installation_id)
    print(token)

register_subcommands(forge_parser)

Register forge subcommands on the given parser.

Called from the main agentic-ci CLI to wire up the forge subcommand group.

Source code in src/agentic_ci/forge/cli.py

def register_subcommands(forge_parser: argparse.ArgumentParser) -> None:
    """Register forge subcommands on the given parser.

    Called from the main ``agentic-ci`` CLI to wire up the ``forge``
    subcommand group.
    """
    forge_parser.add_argument(
        "--token",
        help="GitHub token for authentication (required for GitHub operations)",
    )
    subparsers = forge_parser.add_subparsers(dest="forge_command", required=True)

    p_status = subparsers.add_parser(
        "mr-status", help="Get MR/PR state, source branch, pipeline status"
    )
    p_status.add_argument("url", help="MR/PR URL")
    p_status.set_defaults(func=cmd_mr_status)

    p_comments = subparsers.add_parser("mr-comments", help="Get unresolved review threads")
    p_comments.add_argument("url", help="MR/PR URL")
    p_comments.set_defaults(func=cmd_mr_comments)

    p_general = subparsers.add_parser(
        "mr-general-comments", help="Get general (non-diff-positioned) MR/PR comments"
    )
    p_general.add_argument("url", help="MR/PR URL")
    p_general.add_argument(
        "--since", help="Only return comments created after this ISO 8601 timestamp"
    )
    p_general.set_defaults(func=cmd_mr_general_comments)

    p_reply = subparsers.add_parser("mr-reply", help="Reply to a review thread")
    p_reply.add_argument("url", help="MR/PR URL")
    p_reply.add_argument("thread_id", help="Thread/discussion ID")
    p_reply.add_argument("message", help="Reply message")
    p_reply.set_defaults(func=cmd_mr_reply)

    p_resolve = subparsers.add_parser("mr-resolve", help="Resolve a review thread")
    p_resolve.add_argument("url", help="MR/PR URL")
    p_resolve.add_argument("thread_id", help="Thread/discussion ID")
    p_resolve.set_defaults(func=cmd_mr_resolve)

    p_pipeline = subparsers.add_parser(
        "pipeline-failures", help="Get failed pipeline job names and logs"
    )
    p_pipeline.add_argument("url", help="MR/PR URL")
    p_pipeline.set_defaults(func=cmd_pipeline_failures)

    p_update = subparsers.add_parser("mr-update", help="Update MR/PR title and/or description")
    p_update.add_argument("url", help="MR/PR URL")
    p_update.add_argument("--title", help="New title")
    p_update.add_argument("--description", help="New description")
    p_update.set_defaults(func=cmd_mr_update)

    p_label_exists = subparsers.add_parser(
        "label-exists", help="Check whether a repository/project label exists"
    )
    p_label_exists.add_argument("url", help="Repository URL")
    p_label_exists.add_argument("label", help="Label name")
    p_label_exists.set_defaults(func=cmd_label_exists)

    p_label_create = subparsers.add_parser("label-create", help="Create a repository/project label")
    p_label_create.add_argument("url", help="Repository URL")
    p_label_create.add_argument("label", help="Label name")
    p_label_create.add_argument(
        "--color",
        help="Label color as hex (with or without leading #); default 808080",
    )
    p_label_create.add_argument("--description", help="Label description")
    p_label_create.set_defaults(func=cmd_label_create)

    p_add_labels = subparsers.add_parser("mr-add-labels", help="Attach labels to an existing MR/PR")
    p_add_labels.add_argument("url", help="MR/PR URL")
    p_add_labels.add_argument(
        "--labels",
        nargs="+",
        required=True,
        help="One or more label names to attach",
    )
    p_add_labels.set_defaults(func=cmd_mr_add_labels)

    p_diffpos = subparsers.add_parser(
        "mr-diff-position",
        help="Get first changed line position and diff refs (GitLab only)",
    )
    p_diffpos.add_argument("url", help="MR URL")
    p_diffpos.set_defaults(func=cmd_mr_diff_position)

    p_gh_token = subparsers.add_parser(
        "github-token", help="Generate a GitHub App installation token"
    )
    p_gh_token.add_argument("--app-id", required=True, help="GitHub App ID")
    p_gh_token.add_argument("--installation-id", required=True, help="GitHub App installation ID")
    p_gh_token.add_argument(
        "--private-key", required=True, help="PEM private key string or path to PEM file"
    )
    p_gh_token.set_defaults(func=cmd_github_token)

Session

session

HTTP session and adapter configuration for forge API calls.

Provides auth-injecting adapters for GitLab (PRIVATE-TOKEN) and GitHub (Bearer token), plus a pre-configured session with retry logic.

ForgeAuthError

Bases: RuntimeError

Raised when forge API authentication credentials are missing.

GitLabHTTPAdapter

Bases: HTTPAdapter

Requests adapter that injects PRIVATE-TOKEN for GitLab REST API.

GitHubHTTPAdapter(token, **kwargs)

Bases: HTTPAdapter

Requests adapter that injects Bearer token for GitHub API.

Source code in src/agentic_ci/forge/session.py

def __init__(self, token: str | None, **kwargs):
    self._token = token
    super().__init__(**kwargs)

build_session(*, gitlab_adapter=None, github_token=None)

Build a requests session with forge-specific auth adapters.

The session automatically injects the correct auth headers based on the request URL prefix (gitlab.com or api.github.com).

Parameters:

Name	Type	Description	Default
gitlab_adapter	GitLabHTTPAdapter | None	Custom GitLab HTTP adapter. Defaults to GitLabHTTPAdapter() which uses BOT_PAT.	None
github_token	str | None	Token for GitHub API authentication.	None

Source code in src/agentic_ci/forge/session.py

def build_session(
    *,
    gitlab_adapter: GitLabHTTPAdapter | None = None,
    github_token: str | None = None,
) -> requests.Session:
    """Build a requests session with forge-specific auth adapters.

    The session automatically injects the correct auth headers based
    on the request URL prefix (``gitlab.com`` or ``api.github.com``).

    Args:
        gitlab_adapter: Custom GitLab HTTP adapter. Defaults to
            ``GitLabHTTPAdapter()`` which uses ``BOT_PAT``.
        github_token: Token for GitHub API authentication.
    """
    s = requests.Session()
    s.mount("https://gitlab.com", gitlab_adapter or GitLabHTTPAdapter())
    s.mount(
        "https://api.github.com",
        GitHubHTTPAdapter(token=github_token),
    )
    return s

extract_api_error(resp)

Extract a human-readable error message from a forge API error response.

Tries message, then errors[0].message, falling back to "Unknown error".

Source code in src/agentic_ci/forge/session.py

def extract_api_error(resp: requests.Response) -> str:
    """Extract a human-readable error message from a forge API error response.

    Tries ``message``, then ``errors[0].message``, falling back to
    ``"Unknown error"``.
    """
    try:
        data = resp.json()
        msg = data.get("message") or ""
        if isinstance(msg, list):
            msg = "; ".join(str(m) for m in msg)
        if not msg:
            errors = data.get("errors", [])
            if errors and isinstance(errors[0], dict):
                msg = errors[0].get("message", "")
            elif errors:
                msg = str(errors[0])
        return msg or "Unknown error"
    except (KeyError, TypeError, AttributeError, ValueError):
        return "Unknown error"