tracer-cloud--opensre/tools/cross_vendor/fix_sentry_issue/ship.py

"""Orchestrate shipping a Sentry fix: fresh branch -> commit -> push -> open PR.

Runs only after the coding agent has left a successful fix in the workspace
working tree. Sequences the safe git primitives in :mod:`integrations.git` and the
PR call in :mod:`pr`, and enforces the "never touch the base branch" guarantee: the
fix always lands on a namespaced ``opensre/sentry-fix-*`` branch, the base branch is
never committed to or pushed, and the PR is opened *into* the base branch.

Git primitives raise a neutral :class:`GitCommandError`; this module maps its
``kind`` onto the tool's :class:`FixIssueError` error surface.
"""

from __future__ import annotations

import re
from collections.abc import Mapping, Sequence
from dataclasses import dataclass

from integrations.coding_agent import CodingResult
from integrations.git import (
    GitCommandError,
    changed_paths,
    checkout_branch,
    commit_paths,
    create_branch,
    current_branch,
    default_branch,
    ensure_git_repo,
    file_fingerprints,
    push_branch,
    short_head,
)
from integrations.github.client import resolve_github_token
from tools.cross_vendor.fix_sentry_issue.errors import ERR_NO_CHANGES, ERR_PR_FAILED, FixIssueError
from tools.cross_vendor.fix_sentry_issue.pr import PullRequest, open_pull_request

_BRANCH_PREFIX = "opensre/sentry-fix"
_SUBJECT_MAX = 72


@dataclass(frozen=True)
class ShipResult:
    """Outcome of shipping a fix as a pull request."""

    branch_name: str
    pr: PullRequest


def _slug(issue_id: str) -> str:
    """Filesystem/ref-safe slug for the issue id (falls back to 'issue')."""
    cleaned = re.sub(r"[^A-Za-z0-9_-]+", "-", issue_id).strip("-")
    return cleaned or "issue"


def build_branch_name(workspace: str, issue_id: str) -> str:
    """Namespaced, unique-ish branch name for this fix (never the base branch)."""
    suffix = short_head(workspace) or "wip"
    return f"{_BRANCH_PREFIX}-{_slug(issue_id)}-{suffix}"


def _commit_message(issue_id: str, sentry_url: str, summary: str) -> str:
    subject_body = summary.strip().splitlines()[0] if summary.strip() else "apply proposed fix"
    subject = f"fix: resolve Sentry issue {issue_id} — {subject_body}"[:_SUBJECT_MAX]
    lines = [subject, ""]
    if summary.strip():
        lines += [summary.strip(), ""]
    lines.append(f"Generated by OpenSRE (fix_sentry_issue) from {sentry_url}.")
    return "\n".join(lines)


def _pr_body(issue_id: str, sentry_url: str, result: CodingResult, changed: Sequence[str]) -> str:
    lines = [
        f"Automated fix proposed by OpenSRE for Sentry issue [{issue_id}]({sentry_url}).",
        "",
        "**Summary**",
        result.summary.strip() or "_(no summary provided)_",
    ]
    if changed:
        lines += ["", "**Files changed**"]
        lines += [f"- `{path}`" for path in changed]
    footer = (
        "> Generated by the OpenSRE `fix_sentry_issue` tool (automated coding agent). "
        "Review before merging."
    )
    lines += ["", footer]
    return "\n".join(lines)


def ship_fix(
    workspace: str,
    *,
    issue_id: str,
    sentry_url: str,
    result: CodingResult,
    github_token: str | None = None,
    baseline: Mapping[str, str] | None = None,
) -> ShipResult:
    """Branch, commit, push, and open a PR for the fix in *workspace*.

    Assumes the coding run already succeeded and left changes in the working tree.
    *baseline* is a ``{path: content-hash}`` fingerprint of files already dirty
    **before** the run (see the tool's pre-run snapshot). A dirty file is committed
    only if it is new or its content changed since the baseline — so the agent's
    edits are always shipped (even to a file that was already dirty), while
    pre-existing WIP the agent left untouched is excluded. Raises
    :class:`FixIssueError` (with a stable ``kind``) on any failure.
    """
    # One credential for both the push and the PR call, so the push doesn't fall
    # back to a stale cached git credential (the usual cause of a 403 on push).
    token = resolve_github_token(github_token)

    # Pre-branch phase: nothing has been created yet, so failures carry no branch.
    try:
        ensure_git_repo(workspace)

        # ``result.changed_files`` and a plain worktree scan both report the whole
        # tree vs HEAD, so neither isolates the agent's edits from pre-existing WIP.
        # Compare content hashes against the pre-run baseline: keep files the agent
        # created or actually modified.
        pre_existing = dict(baseline or {})
        current = changed_paths(workspace)
        current_fingerprints = file_fingerprints(workspace, current)
        changed = [
            path
            for path in current
            if path not in pre_existing or current_fingerprints.get(path, "") != pre_existing[path]
        ]
        if not changed:
            reason = (
                "the coding agent made no changes to the repository"
                if not current
                else "the coding agent left no new changes (only pre-existing work-in-progress)"
            )
            raise FixIssueError(
                ERR_NO_CHANGES,
                f"Nothing to commit — {reason}; nothing to open a PR for. "
                "Try re-running or setting a stronger CODING_MODEL.",
            )

        base = default_branch(workspace, token=token or None)
        if not base:
            # Don't guess the base — a PR against the wrong branch is worse than a
            # clear failure. This only happens when origin/HEAD is unset AND the
            # remote is unreachable.
            raise FixIssueError(
                ERR_PR_FAILED,
                "Could not determine the base branch to open the PR against "
                "(origin/HEAD is not configured and the remote is unreachable). "
                "Run `git remote set-head origin -a` in the workspace, or check connectivity.",
            )
        branch = build_branch_name(workspace, issue_id)
        # Branch off the resolved base, never off whatever the workspace happens to
        # have checked out (e.g. an unrelated local feature branch) — unless a prior
        # attempt already left us on this exact fix branch (retry after a partial
        # failure below), in which case re-branching would just fail as "exists".
        if current_branch(workspace) != branch:
            checkout_branch(workspace, base)
            create_branch(workspace, branch, base_default=base)
    except GitCommandError as exc:
        raise FixIssueError(exc.kind, exc.message) from exc

    # From here the workspace is on ``branch``. Any failure (commit, push, or PR)
    # leaves the user's changes on that branch, so tag it onto the error for the
    # caller to surface for manual recovery instead of reporting an empty branch.
    try:
        commit_paths(workspace, changed, _commit_message(issue_id, sentry_url, result.summary))
        push_branch(workspace, branch, base_default=base, token=token or None)
        pr = open_pull_request(
            workspace,
            head_branch=branch,
            base_branch=base,
            title=f"fix: resolve Sentry issue {issue_id}",
            body=_pr_body(issue_id, sentry_url, result, changed),
            github_token=token or None,
        )
    except GitCommandError as exc:
        raise FixIssueError(exc.kind, exc.message, branch_name=branch) from exc
    except FixIssueError as exc:
        exc.branch_name = branch
        raise
    return ShipResult(branch_name=branch, pr=pr)