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")

ForgeError

Bases: Exception

Raised when a forge API operation fails.

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) abstractmethod

Create an MR/PR.

Returns (web_url, None) on success or (None, error_msg) on failure.

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,
) -> tuple[str | None, str | None]:
    """Create an MR/PR.

    Returns ``(web_url, None)`` on success or ``(None, error_msg)``
    on failure.
    """

mr_status(mr_url) abstractmethod

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

Returns {"state": str, "source_branch": str, "pipeline_status": str}. State is normalized to "open", "merged", or "closed".

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

@abstractmethod
def mr_status(self, mr_url: str) -> dict:
    """Get MR/PR state, source branch, and pipeline status.

    Returns ``{"state": str, "source_branch": str, "pipeline_status": str}``.
    State is normalized to ``"open"``, ``"merged"``, or ``"closed"``.
    """

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.

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``.
    """

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. 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``.
    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.
    """

pipeline_failures(mr_url) abstractmethod

Get failed CI job names and log tails.

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

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

@abstractmethod
def pipeline_failures(self, mr_url: str) -> dict:
    """Get failed CI job names and log tails.

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

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()

Bases: Forge

GitLab REST API implementation of the Forge interface.

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

def __init__(self) -> None:
    self._session = build_session()

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"]

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

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 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_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_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(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).

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

def build_session(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``).
    """
    s = requests.Session()
    s.mount("https://gitlab.com", 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"