#!/usr/bin/env python3 """ PR Review Evaluation Script This script runs when a PR is merged or closed to evaluate how well the review comments were addressed. It creates an evaluation trace in Laminar that can be processed by a signal to determine review effectiveness. The evaluation flow: 1. Read the original trace ID from the artifact 2. Fetch PR review comments and thread discussion from GitHub 3. Fetch the final patch/diff 4. Create an evaluation span with all context 5. Optionally score the original trace Environment Variables: LMNR_PROJECT_API_KEY: Laminar project API key (required) GITHUB_TOKEN: GitHub token for API access (required) PR_NUMBER: Pull request number (required) REPO_NAME: Repository name in format owner/repo (required) PR_MERGED: Whether the PR was merged ('true' or 'false') """ import json # Configure logging import logging import os import sys import urllib.error import urllib.request from pathlib import Path from lmnr import Laminar, LaminarClient logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s") logger = logging.getLogger(__name__) FEEDBACK_COMMENT_MARKER = "" REVIEWS_QUERY = """ query($owner: String!, $repo: String!, $pr_number: Int!, $cursor: String) { repository(owner: $owner, name: $repo) { pullRequest(number: $pr_number) { reviews(first: 100, after: $cursor) { pageInfo { hasNextPage endCursor } nodes { id body state submittedAt author { login } reactionGroups { content users { totalCount } } } } } } } """ def _get_required_env(name: str) -> str: """Get a required environment variable or raise an error.""" value = os.getenv(name) if not value: raise ValueError(f"{name} environment variable is required") return value def _get_github_headers() -> dict[str, str]: """Get headers for GitHub API requests.""" token = _get_required_env("GITHUB_TOKEN") return { "Accept": "application/vnd.github.v3+json", "Authorization": f"Bearer {token}", "X-GitHub-Api-Version": "2022-11-28", } def _get_agent_usernames() -> set[str]: """Get the set of agent usernames to identify agent comments. Configurable via AGENT_USERNAMES environment variable (comma-separated). Defaults to 'openhands-agent,all-hands-bot,github-actions[bot]'. """ usernames = os.getenv( "AGENT_USERNAMES", "openhands-agent,all-hands-bot,github-actions[bot]", ) return set(name.strip() for name in usernames.split(",") if name.strip()) def _handle_github_api_error(e: urllib.error.HTTPError, context: str) -> None: """Handle GitHub API errors with rate limit awareness.""" if e.code == 429: retry_after = e.headers.get("Retry-After", "60") logger.warning(f"Rate limited by GitHub API. Retry after {retry_after}s") logger.error(f"Failed to {context}: HTTP {e.code}") def fetch_pr_review_comments(repo: str, pr_number: str) -> list[dict]: """Fetch all review comments on a PR. This includes inline code review comments, not regular PR comments. """ url = f"https://api.github.com/repos/{repo}/pulls/{pr_number}/comments" request = urllib.request.Request(url, headers=_get_github_headers()) try: with urllib.request.urlopen(request, timeout=60) as response: return json.loads(response.read().decode("utf-8")) except urllib.error.HTTPError as e: _handle_github_api_error(e, "fetch review comments") return [] def fetch_pr_issue_comments(repo: str, pr_number: str) -> list[dict]: """Fetch issue-style comments on a PR (the main thread).""" url = f"https://api.github.com/repos/{repo}/issues/{pr_number}/comments" request = urllib.request.Request(url, headers=_get_github_headers()) try: with urllib.request.urlopen(request, timeout=60) as response: return json.loads(response.read().decode("utf-8")) except urllib.error.HTTPError as e: _handle_github_api_error(e, "fetch issue comments") return [] def _call_github_graphql(query: str, variables: dict) -> dict: """Execute a GitHub GraphQL query and return the `data` payload.""" request = urllib.request.Request( "https://api.github.com/graphql", headers=_get_github_headers(), method="POST", data=json.dumps({"query": query, "variables": variables}).encode("utf-8"), ) request.add_header("Content-Type", "application/json") try: with urllib.request.urlopen(request, timeout=60) as response: payload = json.loads(response.read().decode("utf-8")) except urllib.error.HTTPError as e: _handle_github_api_error(e, "fetch GraphQL data") return {} if payload.get("errors"): logger.error("GitHub GraphQL returned errors: %s", payload["errors"]) return {} return payload.get("data") or {} def _normalize_review_reactions(reaction_groups: list[dict] | None) -> dict[str, int]: """Map GraphQL reaction groups to GitHub-style thumbs-up/down counters.""" thumbs_up = 0 thumbs_down = 0 for group in reaction_groups or []: total_count = ((group.get("users") or {}).get("totalCount")) or 0 content = group.get("content") if content == "THUMBS_UP": thumbs_up = total_count elif content == "THUMBS_DOWN": thumbs_down = total_count return { "+1": thumbs_up, "-1": thumbs_down, "total_count": thumbs_up + thumbs_down, } def fetch_pr_reviews(repo: str, pr_number: str) -> list[dict]: """Fetch all reviews on a PR, including thumbs-up/down reaction counts.""" owner, repo_name = repo.split("/", 1) reviews = [] cursor = None while True: data = _call_github_graphql( REVIEWS_QUERY, { "owner": owner, "repo": repo_name, "pr_number": int(pr_number), "cursor": cursor, }, ) reviews_data = ( data.get("repository", {}) .get("pullRequest", {}) .get("reviews", {}) ) nodes = reviews_data.get("nodes") or [] for review in nodes: author = review.get("author") or {} reviews.append( { "id": review.get("id"), "user": {"login": author.get("login")}, "body": review.get("body") or "", "state": review.get("state"), "submitted_at": review.get("submittedAt"), "reactions": _normalize_review_reactions( review.get("reactionGroups") ), } ) page_info = reviews_data.get("pageInfo") or {} if not page_info.get("hasNextPage"): break cursor = page_info.get("endCursor") if not cursor: break return reviews def fetch_pr_diff(repo: str, pr_number: str) -> str: """Fetch the final diff of the PR.""" url = f"https://api.github.com/repos/{repo}/pulls/{pr_number}" headers = _get_github_headers() headers["Accept"] = "application/vnd.github.v3.diff" request = urllib.request.Request(url, headers=headers) try: with urllib.request.urlopen(request, timeout=60) as response: return response.read().decode("utf-8", errors="replace") except urllib.error.HTTPError as e: _handle_github_api_error(e, "fetch PR diff") return "" def fetch_pr_info(repo: str, pr_number: str) -> dict: """Fetch PR metadata.""" url = f"https://api.github.com/repos/{repo}/pulls/{pr_number}" request = urllib.request.Request(url, headers=_get_github_headers()) try: with urllib.request.urlopen(request, timeout=60) as response: return json.loads(response.read().decode("utf-8")) except urllib.error.HTTPError as e: _handle_github_api_error(e, "fetch PR info") return {} def extract_agent_comments( review_comments: list[dict], issue_comments: list[dict], reviews: list[dict] ) -> list[dict]: """Extract comments made by the review agent. Agent usernames are configurable via AGENT_USERNAMES environment variable. """ agent_users = _get_agent_usernames() agent_comments = [] # Review comments (inline code comments) for comment in review_comments: if comment.get("user", {}).get("login") in agent_users: agent_comments.append( { "type": "review_comment", "id": comment.get("id"), "body": comment.get("body", ""), "path": comment.get("path"), "line": comment.get("line") or comment.get("original_line"), "created_at": comment.get("created_at"), } ) # Issue comments (main thread) for comment in issue_comments: if comment.get("user", {}).get("login") in agent_users: agent_comments.append( { "type": "issue_comment", "id": comment.get("id"), "body": comment.get("body", ""), "created_at": comment.get("created_at"), } ) # Review bodies for review in reviews: if review.get("user", {}).get("login") in agent_users and review.get("body"): agent_comments.append( { "type": "review", "id": review.get("id"), "body": review.get("body", ""), "state": review.get("state"), "created_at": review.get("submitted_at"), } ) return agent_comments def extract_human_responses( review_comments: list[dict], issue_comments: list[dict], agent_users: set[str] | None = None, ) -> list[dict]: """Extract comments/responses from humans (non-agent users). Agent usernames are configurable via AGENT_USERNAMES environment variable. """ if agent_users is None: agent_users = _get_agent_usernames() human_responses = [] for comment in review_comments: if comment.get("user", {}).get("login") not in agent_users: human_responses.append( { "type": "review_comment", "user": comment.get("user", {}).get("login"), "body": comment.get("body", ""), "in_reply_to_id": comment.get("in_reply_to_id"), "created_at": comment.get("created_at"), } ) for comment in issue_comments: if comment.get("user", {}).get("login") not in agent_users: human_responses.append( { "type": "issue_comment", "user": comment.get("user", {}).get("login"), "body": comment.get("body", ""), "created_at": comment.get("created_at"), } ) return human_responses def extract_review_feedback( issue_comments: list[dict], reviews: list[dict] | None = None ) -> list[dict]: """Extract thumbs-up/down feedback from review bodies or legacy comments.""" agent_users = _get_agent_usernames() feedback = [] for comment in [*issue_comments, *(reviews or [])]: if FEEDBACK_COMMENT_MARKER not in (comment.get("body") or ""): continue if comment.get("user", {}).get("login") not in agent_users: continue reactions = comment.get("reactions") or {} thumbs_up = reactions.get("+1", 0) or 0 thumbs_down = reactions.get("-1", 0) or 0 feedback.append( { "comment_id": comment.get("id"), "created_at": comment.get("created_at") or comment.get("submitted_at"), "thumbs_up": thumbs_up, "thumbs_down": thumbs_down, "total": thumbs_up + thumbs_down, } ) return feedback def truncate_text(text: str, max_chars: int = 50000) -> str: """Truncate text to stay within reasonable API payload limits. Max 50k chars chosen to stay well under typical API payload limits while preserving enough context for evaluation. This keeps the evaluation trace size manageable for Laminar processing. """ if len(text) <= max_chars: return text return text[:max_chars] + f"\n\n... [truncated, {len(text)} total chars]" def load_trace_info(trace_file_path: str | None = None) -> dict: """Load trace info from artifact file. Args: trace_file_path: Path to trace info JSON file. If None, uses default path. Returns: Dictionary with trace_id, span_context, and other metadata. Empty dict if file not found. """ trace_info_path = Path(trace_file_path) if trace_file_path else Path("laminar_trace_info.json") if not trace_info_path.exists(): logger.warning( "No trace info file found - evaluation will create standalone trace" ) return {} with open(trace_info_path) as f: data = json.load(f) logger.info(f"Original trace ID: {data.get('trace_id')}") if data.get("span_context"): logger.info("Found span context - will add evaluation to original trace") else: logger.info("No span context - evaluation will create standalone trace") return data def fetch_pr_data(repo: str, pr_number: str) -> dict: """Fetch all PR data from GitHub. Args: repo: Repository in format owner/repo pr_number: PR number Returns: Dictionary with review_comments, issue_comments, reviews, final_diff, pr_info, agent_comments, and human_responses """ logger.info("Fetching PR data from GitHub...") review_comments = fetch_pr_review_comments(repo, pr_number) issue_comments = fetch_pr_issue_comments(repo, pr_number) reviews = fetch_pr_reviews(repo, pr_number) final_diff = fetch_pr_diff(repo, pr_number) pr_info = fetch_pr_info(repo, pr_number) logger.info(f"Found {len(review_comments)} review comments") logger.info(f"Found {len(issue_comments)} issue comments") logger.info(f"Found {len(reviews)} reviews") agent_comments = extract_agent_comments(review_comments, issue_comments, reviews) human_responses = extract_human_responses(review_comments, issue_comments) review_feedback = extract_review_feedback(issue_comments, reviews) logger.info(f"Agent made {len(agent_comments)} comments") logger.info(f"Humans made {len(human_responses)} responses") logger.info(f"Found {len(review_feedback)} review feedback prompts") return { "review_comments": review_comments, "issue_comments": issue_comments, "reviews": reviews, "final_diff": final_diff, "pr_info": pr_info, "agent_comments": agent_comments, "human_responses": human_responses, "review_feedback": review_feedback, } def calculate_engagement_score( agent_comments: list[dict], human_responses: list[dict], pr_merged: bool, ) -> float: """Calculate engagement score based on interaction metrics. Components: - Response ratio: humans responded to agent comments (0-0.5) - Completion bonus: PR was merged (0.3) Max score: 0.8 Args: agent_comments: List of agent comments human_responses: List of human responses pr_merged: Whether the PR was merged Returns: Engagement score between 0.0 and 0.8 """ score = 0.0 if agent_comments: engagement_ratio = min(len(human_responses) / len(agent_comments), 1.0) score = engagement_ratio * 0.5 if pr_merged: score += 0.3 return score def create_evaluation_span( pr_number: str, repo_name: str, pr_merged: bool, pr_data: dict, trace_info: dict, ) -> str | None: """Create Laminar evaluation span and return trace ID. Args: pr_number: PR number repo_name: Repository name pr_merged: Whether PR was merged pr_data: Dictionary from fetch_pr_data() trace_info: Dictionary from load_trace_info() Returns: Evaluation trace ID, or None if not available """ Laminar.initialize() evaluation_context = { "pr_number": pr_number, "repo_name": repo_name, "pr_merged": pr_merged, "pr_title": pr_data["pr_info"].get("title", ""), "pr_state": pr_data["pr_info"].get("state", ""), "original_trace_id": trace_info.get("trace_id"), "agent_comments": pr_data["agent_comments"], "human_responses": pr_data["human_responses"], "review_feedback": pr_data["review_feedback"], "final_diff": truncate_text(pr_data["final_diff"]), "total_review_comments": len(pr_data["review_comments"]), "total_issue_comments": len(pr_data["issue_comments"]), } with Laminar.start_as_current_span( name="pr_review_evaluation", input=evaluation_context, tags=["pr-review-evaluation"], parent_span_context=trace_info.get("span_context"), ): Laminar.set_trace_metadata( { "original_trace_id": trace_info.get("trace_id") or "none", "evaluation_type": "pr_review_effectiveness", "pr_number": pr_number, "repo_name": repo_name, "pr_merged": str(pr_merged), } ) summary = { "pr": f"{repo_name}#{pr_number}", "merged": pr_merged, "agent_comments_count": len(pr_data["agent_comments"]), "human_responses_count": len(pr_data["human_responses"]), "review_feedback": pr_data["review_feedback"], "diff_length": len(pr_data["final_diff"]), } logger.info(f"Evaluation summary: {json.dumps(summary)}") Laminar.set_span_output( { "summary": summary, "ready_for_signal": True, } ) eval_trace_id = Laminar.get_trace_id() Laminar.flush() return str(eval_trace_id) if eval_trace_id else None def main(trace_file_path: str | None = None): """Run the PR review evaluation. Args: trace_file_path: Optional path to trace info JSON file. """ logger.info("Starting PR review evaluation...") pr_number = _get_required_env("PR_NUMBER") repo_name = _get_required_env("REPO_NAME") pr_merged = os.getenv("PR_MERGED", "false").lower() == "true" logger.info(f"Evaluating PR #{pr_number} in {repo_name}") logger.info(f"PR was merged: {pr_merged}") trace_info = load_trace_info(trace_file_path) pr_data = fetch_pr_data(repo_name, pr_number) eval_trace_id = create_evaluation_span( pr_number, repo_name, pr_merged, pr_data, trace_info ) original_trace_id = trace_info.get("trace_id") agent_comments = pr_data["agent_comments"] human_responses = pr_data["human_responses"] review_feedback = pr_data["review_feedback"] # Score engagement on the original trace for immediate feedback if original_trace_id: try: client = LaminarClient() engagement_score = calculate_engagement_score( agent_comments, human_responses, pr_merged ) client.evaluators.score( name="review_engagement", trace_id=original_trace_id, score=engagement_score, metadata={ "agent_comments": len(agent_comments), "human_responses": len(human_responses), "pr_merged": pr_merged, "review_feedback": review_feedback, "score_type": "engagement", }, ) logger.info( f"Added engagement score {engagement_score:.2f} " f"to original trace {original_trace_id}" ) client.tags.tag(original_trace_id, ["evaluated", f"pr-{pr_number}"]) logger.info(f"Tagged original trace {original_trace_id}") except Exception as e: logger.warning(f"Failed to score original trace: {e}") # Print evaluation summary print("\n=== PR Review Evaluation ===") print(f"PR: {repo_name}#{pr_number}") print(f"Merged: {pr_merged}") print(f"Agent Comments: {len(agent_comments)}") print(f"Human Responses: {len(human_responses)}") if review_feedback: thumbs_up = sum(item["thumbs_up"] for item in review_feedback) thumbs_down = sum(item["thumbs_down"] for item in review_feedback) print(f"Review Feedback: 👍 {thumbs_up} / 👎 {thumbs_down}") if original_trace_id: print(f"Original Review Trace: {original_trace_id}") if eval_trace_id: print(f"Evaluation Trace: {eval_trace_id}") logger.info("PR review evaluation completed successfully") if __name__ == "__main__": import argparse parser = argparse.ArgumentParser(description="Evaluate PR review effectiveness") parser.add_argument( "--trace-file", help="Path to trace info JSON file (default: laminar_trace_info.json)", ) args = parser.parse_args() try: main(trace_file_path=args.trace_file) except Exception as e: logger.error(f"Evaluation failed: {e}") sys.exit(1)