Merge pull request #61 from Contrast-Security-OSS/AIML-150_fix_finding_pr_for_copilot

Aiml 150 fix finding pr for copilot

Author: JacobMagesHaskinsContrast

src/git_handler.py

@@ -625,7 +625,7 @@ def find_issue_with_label(label: str) -> int:
         return None
 
 
-def reset_issue(issue_number: int, remediation_label: str) -> bool:
+def reset_issue(issue_number: int, issue_title: str, remediation_label: str) -> bool:
     """
     Resets a GitHub issue by:
     1. Removing all existing labels that start with "smartfix-id:"
@@ -650,7 +650,7 @@ def reset_issue(issue_number: int, remediation_label: str) -> bool:
         return False
 
     # First check if there's an open PR for this issue
-    open_pr = find_open_pr_for_issue(issue_number)
+    open_pr = find_open_pr_for_issue(issue_number, issue_title)
     if open_pr:
         pr_number = open_pr.get("number")
         pr_url = open_pr.get("url")
@@ -759,7 +759,7 @@ def reset_issue(issue_number: int, remediation_label: str) -> bool:
         return False
 
 
-def find_open_pr_for_issue(issue_number: int) -> dict:
+def find_open_pr_for_issue(issue_number: int, issue_title: str) -> dict:
     """
     Finds an open pull request associated with the given issue number.
     Specifically looks for PRs with branch names matching the pattern 'copilot/fix-{issue_number}'
@@ -805,8 +805,22 @@ def find_open_pr_for_issue(issue_number: int) -> dict:
             pr_list_output = run_command(claude_pr_list_command, env=gh_env, check=False)
 
             if not pr_list_output or pr_list_output.strip() == "[]":
-                debug_log(f"No open PRs found for issue #{issue_number} with either Copilot or Claude branch pattern")
-                return None
+                escaped_issue_title = issue_title.replace('"', '\\"')
+                copilot_issue_title_search_pattern = f"in:title \"[WIP] {escaped_issue_title}\""
+                copilot_issue_title_list_command = [
+                    "gh", "pr", "list",
+                    "--repo", config.GITHUB_REPOSITORY,
+                    "--state", "open",
+                    "--search", copilot_issue_title_search_pattern,
+                    "--limit", "1",
+                    "--json", "number,url,title,headRefName,baseRefName,state"
+                ]
+
+                pr_list_output = run_command(copilot_issue_title_list_command, env=gh_env, check=False)
+
+                if not pr_list_output or pr_list_output.strip() == "[]":
+                    debug_log(f"No open PRs found for issue #{issue_number} with either Copilot or Claude branch pattern")
+                    return None
 
         prs_data = json.loads(pr_list_output)
 
@@ -926,7 +940,7 @@ def get_issue_comments(issue_number: int, author: str = None) -> List[dict]:
     Returns:
         List[dict]: A list of comment data dictionaries or empty list if no comments or error
     """
-    author_log = f"and author: {author}" if author  else ""
+    author_log = f"and author: {author}" if author else ""
     debug_log(f"Getting comments for issue #{issue_number} {author_log}")
     gh_env = get_gh_env()
     author_filter = f"| map(select(.author.login == \"{author}\")) " if author else ""
@@ -1098,7 +1112,11 @@ def get_claude_workflow_run_id() -> int:
     debug_log("Getting in-progress Claude workflow run ID")
 
     gh_env = get_gh_env()
-    jq_filter = 'map(select(.event == "issues" or .event == "issue_comment") | select(.status == "in_progress") | select(.conclusion != "skipped")) | sort_by(.createdAt) | reverse | .[0]'
+    jq_filter = (
+        'map(select(.event == "issues" or .event == "issue_comment") | '
+        'select(.status == "in_progress") | select(.conclusion != "skipped")) | '
+        'sort_by(.createdAt) | reverse | .[0]'
+    )
     workflow_command = [
         "gh", "run", "list",
         "--repo", config.GITHUB_REPOSITORY,

src/github/external_coding_agent.py

@@ -183,7 +183,7 @@ def generate_fixes(self, vuln_uuid: str, remediation_id: str, vuln_title: str, i
                 error_exit(remediation_id, FailureCategory.AGENT_FAILURE.value)
         else:
             debug_log(f"Found existing GitHub issue #{issue_number} with label {vulnerability_label}")
-            if not git_handler.reset_issue(issue_number, remediation_label):
+            if not git_handler.reset_issue(issue_number, issue_title, remediation_label):
                 log(f"Failed to reset issue #{issue_number} with labels {vulnerability_label}, {remediation_label}", is_error=True)
                 error_exit(remediation_id, FailureCategory.AGENT_FAILURE.value)
             is_existing_issue = True
@@ -193,10 +193,13 @@ def generate_fixes(self, vuln_uuid: str, remediation_id: str, vuln_title: str, i
         # Proceed with PR polling for all agent types
 
         # Poll for PR creation by the external agent
-        log(f"Waiting for external agent to create a PR for issue #{issue_number}")
+        log(f"Waiting for external agent to create a PR for issue #{issue_number}, '{issue_title}'")
 
         # Poll for a PR to be created by the external agent (100 attempts, 5 seconds apart = ~8.3 minutes max)
-        pr_info = self._process_external_coding_agent_run(issue_number, remediation_id, vulnerability_label, remediation_label, is_existing_issue, max_attempts=100, sleep_seconds=5)
+        pr_info = self._process_external_coding_agent_run(
+            issue_number, issue_title, remediation_id, vulnerability_label,
+            remediation_label, is_existing_issue, max_attempts=100, sleep_seconds=5
+        )
 
         log("\n::endgroup::")
         if pr_info:
@@ -215,7 +218,7 @@ def generate_fixes(self, vuln_uuid: str, remediation_id: str, vuln_title: str, i
             telemetry_handler.update_telemetry("resultInfo.failureCategory", FailureCategory.AGENT_FAILURE.name)
             return False
 
-    def _process_external_coding_agent_run(self, issue_number: int, remediation_id: str, vulnerability_label: str,
+    def _process_external_coding_agent_run(self, issue_number: int, issue_title: str, remediation_id: str, vulnerability_label: str,
                                            remediation_label: str, is_existing_issue: bool,
                                            max_attempts: int = 100, sleep_seconds: int = 5) -> Optional[dict]:
         """
@@ -246,7 +249,7 @@ def _process_external_coding_agent_run(self, issue_number: int, remediation_id:
                 pr_info = self._process_claude_workflow_run(issue_number, remediation_id)
             else:
                 # GitHub Copilot agent
-                pr_info = git_handler.find_open_pr_for_issue(issue_number)
+                pr_info = git_handler.find_open_pr_for_issue(issue_number, issue_title)
 
             if pr_info:
                 pr_number = pr_info.get("number")
@@ -288,7 +291,6 @@ def _process_external_coding_agent_run(self, issue_number: int, remediation_id:
         log(f"No PR found for issue #{issue_number} after {max_attempts} polling attempts", is_error=True)
         return None
 
-
     def _process_claude_workflow_run(self, issue_number: int, remediation_id: str,) -> Optional[dict]:
         """
         Process the Claude Code workflow run and extract PR information from Claude's comment
@@ -307,13 +309,13 @@ def _process_claude_workflow_run(self, issue_number: int, remediation_id: str,)
 
             if not workflow_run_id:
                 # If no workflow run ID found yet, continue polling
-                debug_log(f"Claude workflow_run_id not found, checking again...")
+                debug_log("Claude workflow_run_id not found, checking again...")
                 return None
 
             # Get all issue comments to find the latest comment author.login
             issue_comments = git_handler.get_issue_comments(issue_number)
             if not issue_comments or len(issue_comments) == 0:
-                debug_log(f"No comments added to issue, checking again...")
+                debug_log("No comments added to issue, checking again...")
                 return None
 
             author_login = issue_comments[0].get("author", {}).get("login", '')
@@ -364,8 +366,8 @@ def _process_claude_workflow_run(self, issue_number: int, remediation_id: str,)
             debug_log(f"Claude create PR returned url: {pr_url}")
 
             if not pr_url:
-                log(f"Failed to create PR for Claude Code fix", is_error=True)
-                reason = f"Could not create Claude PR due to processing issues"
+                log("Failed to create PR for Claude Code fix", is_error=True)
+                reason = "Could not create Claude PR due to processing issues"
                 self._update_telemetry_and_exit_claude_agent_failure(reason, remediation_id, issue_number)
 
             # Extract PR number from URL
@@ -396,8 +398,7 @@ def _process_claude_workflow_run(self, issue_number: int, remediation_id: str,)
             self._update_telemetry_and_exit_claude_agent_failure(msg, remediation_id, issue_number)
             return None
 
-
-    def _process_claude_comment_body(self, comment_body: str, remediation_id: str, issue_number: int) -> dict:
+    def _process_claude_comment_body(self, comment_body: str, remediation_id: str, issue_number: int) -> dict:  # noqa: C901
         """
         Process Claude's comment body to extract PR information. Returning the pr_title
         and pr_body are required for this method to be successful and to create the PR.
@@ -499,7 +500,6 @@ def _process_claude_comment_body(self, comment_body: str, remediation_id: str, i
             "pr_body": contrast_pr_body
         }
 
-
     def _get_claude_head_branch(self, head_branch_from_url: str,
                                 comment_body: str,
                                 issue_number: int,
@@ -547,8 +547,8 @@ def _get_claude_head_branch(self, head_branch_from_url: str,
 
         # Final check - if no branch could be found by any method, fail gracefully
         if not head_branch:
-            log(f"Could not determine claude branch name using any available method", is_error=True)
-            reason = f"Could not extract Claude head_branch needed for PR creation"
+            log("Could not determine claude branch name using any available method", is_error=True)
+            reason = "Could not extract Claude head_branch needed for PR creation"
             self._update_telemetry_and_exit_claude_agent_failure(reason, remediation_id, issue_number)
 
         return head_branch