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