PR Review Flow

async def process_pr_review(self, payload: dict, event: Event, db: AsyncSession) -> dict:
    pr = payload["pull_request"]
    repo_full_name = payload.get("repository", {}).get("full_name", "")
    pr_number = pr["number"]
    head_sha = (pr.get("head") or {}).get("sha", "")

{"status": "completed", "summary": "...", "files_analyzed": 12, "inline_suggestions": 3}

diff = await github_client.get_pr_diff(owner, repo, pr_number)
files = await github_client.get_pr_files(owner, repo, pr_number)

issue_details, open_pr_conflicts, candidate_issues, related_prs = await asyncio.gather(
    _fetch_issue_details(owner, repo, issue_refs),
    _get_open_pr_conflicts(owner, repo, pr_number, file_paths),
    _find_candidate_issues(owner, repo, pr_title, pr_body, file_paths, already_referenced=set(issue_refs)),
    graph_builder.get_related_prs(repo_full_name, file_paths[:10], top_k=5),
    return_exceptions=True,
)

_ISSUE_REF_PATTERN = re.compile(r"(?:^|(?<=\s))(?:fixes|closes|resolves)\s+#(\d+)", re.IGNORECASE | re.MULTILINE)

REVIEW_TOOLS = [
    {"name": "read_file", "description": "Read complete source at head commit"},
    {"name": "search_project_memory", "description": "Search Mem0 for patterns/decisions"},
    {"name": "search_developer_memory", "description": "Author-specific patterns"},
    {"name": "get_file_history", "description": "File experts + past PRs"},
    {"name": "get_issue_details", "description": "Fetch GitHub issue details"},
    {"name": "search_open_issues", "description": "Keyword search in open issues"},
    {"name": "get_linked_issues", "description": "Linear/GitHub issue search"},
    {"name": "get_related_errors", "description": "Sentry errors for modified files"},
]

class ReviewToolExecutor:
    def __init__(self, owner, repo, repo_full_name, head_sha, author, candidate_issues):
        self.owner = owner
        self.repo = repo
        self.head_sha = head_sha
        self.candidate_issues = candidate_issues

    async def execute(self, tool_name: str, tool_input: dict) -> str:
        if tool_name == "read_file":
            return await self._read_file(tool_input["path"])
        if tool_name == "search_project_memory":
            return await self._search_project_memory(tool_input["query"])
        # ... 8 tools total

for round_num in range(max_rounds):  # max_rounds = 8
    response = await self.client.messages.create(
        model=self.model,
        max_tokens=4000,
        tools=REVIEW_TOOLS,
        messages=messages,
    )

    if response.stop_reason == "end_turn":
        raw_text = "".join(b.text for b in response.content if hasattr(b, "text"))
        break

    if response.stop_reason == "tool_use":
        # Execute tool calls → append results → continue loop
        for block in response.content:
            if block.type == "tool_use":
                result = await tool_executor.execute(block.name, block.input)
                tool_results.append({"type": "tool_result", "tool_use_id": block.id, "content": result})
        messages.append({"role": "user", "content": tool_results})
        continue

# Claude outputs:
# 1. Prose review (markdown)
# 2. <suggestions>[{file, line_hint, comment, suggestion}]</suggestions>
# 3. <semantic_issues>[{number, confidence, reason}]</semantic_issues>

suggestions_match = _SUGGESTIONS_RE.search(raw_text)
if suggestions_match:
    inline_comments = json.loads(suggestions_match.group(1))[:5]

semantic_match = _SEMANTIC_ISSUES_RE.search(raw_text)
if semantic_match:
    semantic_issue_matches = json.loads(semantic_match.group(1))[:5]

def _build_line_map(files: list[dict]) -> dict[str, dict[str, int]]:
    """Parse diff patches to map line content → right-side line number"""
    line_map = {}
    for f in files:
        patch = f.get("patch", "")
        mapping = {}
        for patch_line in patch.splitlines():
            if patch_line.startswith("@@"):
                current_right_line = int(_HUNK_HEADER_RE.search(patch_line).group(1)) - 1
            elif patch_line.startswith("+"):
                current_right_line += 1
                content = patch_line[1:]  # strip '+'
                mapping[content.strip()] = current_right_line
        line_map[filename] = mapping
    return line_map

def _resolve_line(hint: str, lines: dict[str, int]) -> int | None:
    return lines.get(hint) or lines.get(hint.strip()) or lines.get(_normalize_ws(hint.strip()))

# Map AI verdict to GitHub review event
_event_map = {
    "APPROVE": "APPROVE",
    "REQUEST_CHANGES": "REQUEST_CHANGES",
    "NEEDS_DISCUSSION": "COMMENT",
}
github_event = _event_map.get(review_result.verdict, "COMMENT")

await github_client.post_pr_review(
    owner, repo, pr_number,
    commit_id=head_sha,
    body=comment_body,
    event=github_event,
    comments=inline_comments,  # [{path, line, side, body}]
)

await graph_builder.index_pr(
    repo_full_name=repo_full_name,
    pr_number=pr_number,
    title=pr_title,
    author=author,
    files_changed=file_paths,
    verdict=review_result.verdict,
    issue_numbers=all_issue_numbers,
)

await extract_and_store(
    repo_full_name=repo_full_name,
    pr_number=pr_number,
    author=author,
    title=pr_title,
    files=files,
    review_summary=summary,
)

except Exception as e:
    logger.error(f"PR review failed for {repo_full_name}#{pr_number}: {e}", exc_info=True)
    workflow.status = "failed"
    workflow.error = str(e)
    event.status = "failed"
    return {"status": "failed", "error": str(e)}

_SKIP_FILE_NAMES = {"package-lock.json", "yarn.lock", "pnpm-lock.yaml", "poetry.lock", "Cargo.lock"}
_SKIP_FILE_EXTS = {".min.js", ".min.css", ".map", ".snap", ".lock", ".pb", ".pyc"}