mirror of
https://github.com/Sea-Haven-Industries/open-swe.git
synced 2026-09-30 22:03:14 +00:00
8d52e17614
When `_ensure_repo_checked_out` or `compute_diff_in_sandbox` fail, the reviewer used to log+continue and hand the agent an empty diff. The agent would then emit a misleading "no issues found" review and the underlying error stayed buried in server logs. Now the helpers raise on non-zero exit codes (with the sandbox output) and the prep block re-raises, so the failure shows up in the LangSmith run trace and we can debug the real issue.
239 lines
8.3 KiB
Python
239 lines
8.3 KiB
Python
"""Diff utilities for the reviewer agent.
|
|
|
|
The reviewer needs three things from a PR diff:
|
|
|
|
1. The set of (file, line) tuples that are part of the diff, so ``add_finding``
|
|
can validate at creation time rather than at GitHub-publish time.
|
|
2. The hunk text relevant to a given (file, start_line, end_line) range, so we
|
|
can stash it on the Finding (``diff_hunk``) for rendering in the future UI
|
|
without re-fetching from GitHub or the (evictable) sandbox.
|
|
3. A way to compute the diff in the sandbox between two SHAs, used both on
|
|
first review (``base_sha..head_sha``) and on watched re-review
|
|
(``last_reviewed_sha..new_head_sha``).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
import re
|
|
from dataclasses import dataclass
|
|
from typing import TYPE_CHECKING
|
|
|
|
if TYPE_CHECKING:
|
|
from deepagents.backends.protocol import SandboxBackendProtocol
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
_DIFF_FILE_HEADER_RE = re.compile(r"^diff --git a/(?P<a>.+?) b/(?P<b>.+?)$")
|
|
_HUNK_HEADER_RE = re.compile(
|
|
r"^@@ -(?P<old_start>\d+)(?:,(?P<old_count>\d+))? "
|
|
r"\+(?P<new_start>\d+)(?:,(?P<new_count>\d+))? @@"
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class DiffHunk:
|
|
"""One hunk for one file in a unified diff.
|
|
|
|
``new_start``/``new_end`` are inclusive 1-based line numbers in the
|
|
post-PR (RIGHT side) file. ``body`` is the raw hunk text including the
|
|
``@@`` header — what gets stored on a Finding's ``diff_hunk``.
|
|
"""
|
|
|
|
file: str
|
|
new_start: int
|
|
new_end: int
|
|
old_start: int
|
|
old_end: int
|
|
body: str
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class FileDiff:
|
|
"""All hunks for one file in a unified diff."""
|
|
|
|
file: str
|
|
hunks: tuple[DiffHunk, ...]
|
|
|
|
|
|
def parse_unified_diff(diff_text: str) -> list[FileDiff]:
|
|
"""Parse a unified diff into per-file hunk records.
|
|
|
|
Skips ``--- ``/``+++ `` and binary file markers. Returns one ``FileDiff``
|
|
per file with at least one hunk; files with no hunks (e.g., pure renames)
|
|
are dropped.
|
|
"""
|
|
files: list[FileDiff] = []
|
|
lines = diff_text.splitlines()
|
|
i = 0
|
|
while i < len(lines):
|
|
header_match = _DIFF_FILE_HEADER_RE.match(lines[i])
|
|
if not header_match:
|
|
i += 1
|
|
continue
|
|
file_path = header_match.group("b")
|
|
i += 1
|
|
# Skip metadata lines until first hunk or next file header
|
|
hunks: list[DiffHunk] = []
|
|
current_hunk_lines: list[str] = []
|
|
current_meta: tuple[int, int, int, int] | None = None
|
|
while i < len(lines) and not _DIFF_FILE_HEADER_RE.match(lines[i]):
|
|
line = lines[i]
|
|
hunk_match = _HUNK_HEADER_RE.match(line)
|
|
if hunk_match:
|
|
if current_meta is not None and current_hunk_lines:
|
|
hunks.append(
|
|
DiffHunk(
|
|
file=file_path,
|
|
old_start=current_meta[0],
|
|
old_end=current_meta[1],
|
|
new_start=current_meta[2],
|
|
new_end=current_meta[3],
|
|
body="\n".join(current_hunk_lines),
|
|
)
|
|
)
|
|
old_start = int(hunk_match.group("old_start"))
|
|
old_count = int(hunk_match.group("old_count") or "1")
|
|
new_start = int(hunk_match.group("new_start"))
|
|
new_count = int(hunk_match.group("new_count") or "1")
|
|
# End line is inclusive; if count is 0 (deletion-only), end == start
|
|
old_end = old_start + max(old_count - 1, 0)
|
|
new_end = new_start + max(new_count - 1, 0)
|
|
current_meta = (old_start, old_end, new_start, new_end)
|
|
current_hunk_lines = [line]
|
|
elif current_meta is not None:
|
|
current_hunk_lines.append(line)
|
|
i += 1
|
|
if current_meta is not None and current_hunk_lines:
|
|
hunks.append(
|
|
DiffHunk(
|
|
file=file_path,
|
|
old_start=current_meta[0],
|
|
old_end=current_meta[1],
|
|
new_start=current_meta[2],
|
|
new_end=current_meta[3],
|
|
body="\n".join(current_hunk_lines),
|
|
)
|
|
)
|
|
if hunks:
|
|
files.append(FileDiff(file=file_path, hunks=tuple(hunks)))
|
|
return files
|
|
|
|
|
|
def compute_diff_line_set(diff_text: str) -> dict[str, set[int]]:
|
|
"""Return ``{file: {line, ...}}`` for the new-side lines covered by the diff.
|
|
|
|
A ``Finding`` whose ``(file, start_line..end_line)`` range falls outside
|
|
this set cannot be rendered as an inline GitHub review comment, so
|
|
``add_finding`` rejects it.
|
|
"""
|
|
out: dict[str, set[int]] = {}
|
|
for file_diff in parse_unified_diff(diff_text):
|
|
lines = out.setdefault(file_diff.file, set())
|
|
for hunk in file_diff.hunks:
|
|
for line in range(hunk.new_start, hunk.new_end + 1):
|
|
lines.add(line)
|
|
return out
|
|
|
|
|
|
def extract_diff_hunk(
|
|
diff_text: str,
|
|
file: str,
|
|
start_line: int | None,
|
|
end_line: int | None,
|
|
) -> str | None:
|
|
"""Extract the hunk body covering ``file:start_line..end_line``.
|
|
|
|
Returns ``None`` if no hunk overlaps. For file-level findings (both lines
|
|
None) returns the first hunk in the file as best-effort context.
|
|
"""
|
|
file_diffs = [fd for fd in parse_unified_diff(diff_text) if fd.file == file]
|
|
if not file_diffs:
|
|
return None
|
|
hunks = file_diffs[0].hunks
|
|
if not hunks:
|
|
return None
|
|
if start_line is None or end_line is None:
|
|
return hunks[0].body
|
|
for hunk in hunks:
|
|
if hunk.new_start <= end_line and start_line <= hunk.new_end:
|
|
return hunk.body
|
|
return None
|
|
|
|
|
|
def is_range_in_diff(
|
|
line_set: dict[str, set[int]],
|
|
file: str,
|
|
start_line: int | None,
|
|
end_line: int | None,
|
|
) -> bool:
|
|
"""Return True if every line in ``start_line..end_line`` is in the diff.
|
|
|
|
File-level findings (both None) are always allowed.
|
|
"""
|
|
if start_line is None and end_line is None:
|
|
return True
|
|
if start_line is None or end_line is None:
|
|
return False
|
|
file_lines = line_set.get(file)
|
|
if not file_lines:
|
|
return False
|
|
return all(line in file_lines for line in range(start_line, end_line + 1))
|
|
|
|
|
|
async def compute_diff_in_sandbox(
|
|
sandbox_backend: SandboxBackendProtocol,
|
|
work_dir: str,
|
|
base_ref: str,
|
|
head_ref: str,
|
|
*,
|
|
merge_base: bool = False,
|
|
) -> str:
|
|
"""Run ``git diff`` inside the sandbox and return its stdout.
|
|
|
|
Refs can be SHAs or branch names. Caller is responsible for ensuring both
|
|
refs exist locally (e.g., having fetched the PR head).
|
|
|
|
Args:
|
|
merge_base: When ``True``, use three-dot ``base...head`` (the merge-base
|
|
diff — what GitHub shows on the PR's "Files changed" tab). Use this
|
|
for first review so we don't pick up changes that landed on the
|
|
base branch after the PR diverged. When ``False``, use two-dot
|
|
``base..head`` — appropriate for re-review deltas where ``base`` is
|
|
the previously reviewed SHA and we want exactly the commits added
|
|
since.
|
|
"""
|
|
operator = "..." if merge_base else ".."
|
|
cmd = f"cd {work_dir} && git diff --no-color {base_ref}{operator}{head_ref}"
|
|
result = await asyncio.to_thread(sandbox_backend.execute, cmd)
|
|
exit_code = getattr(result, "exit_code", None)
|
|
if exit_code not in (0, None):
|
|
output = _stdout_from_result(result)
|
|
raise RuntimeError(
|
|
f"git diff failed (exit {exit_code}) for "
|
|
f"{base_ref}{operator}{head_ref} in {work_dir}. Output:\n{output}"
|
|
)
|
|
return _stdout_from_result(result)
|
|
|
|
|
|
def _stdout_from_result(result: object) -> str:
|
|
"""Best-effort extraction of stdout from a sandbox execute() result.
|
|
|
|
Different sandbox providers return different shapes; this normalizes them.
|
|
"""
|
|
if isinstance(result, str):
|
|
return result
|
|
if isinstance(result, dict):
|
|
for key in ("stdout", "output", "text"):
|
|
value = result.get(key)
|
|
if isinstance(value, str):
|
|
return value
|
|
stdout = getattr(result, "stdout", None)
|
|
if isinstance(stdout, str):
|
|
return stdout
|
|
text = getattr(result, "text", None)
|
|
if isinstance(text, str):
|
|
return text
|
|
return ""
|