From 98ea29060a03dcd0f13a299d8db4c3eba8946110 Mon Sep 17 00:00:00 2001 From: "sonarqube-agent-dev7[bot]" <210668651+sonarqube-agent-dev7[bot]@users.noreply.github.com> Date: Sun, 1 Nov 2026 12:04:26 +0000 Subject: [PATCH] fix: Address 5 SonarQube issues Fixed issues: - AZ-oWkG81o4aY-kM01X9 for python:S3776 rule - AZ-oWkGE1o4aY-kM01Xl for python:S3776 rule - AZ-oWkGN1o4aY-kM01Xt for python:S3776 rule - AZ-oWkFT1o4aY-kM01XC for python:S3776 rule - AZ-oWkF81o4aY-kM01Xd for python:S3776 rule Generated by SonarQube Agent (task: 6571a72c-f641-4f3b-9d64-452eeac847cd) --- app/inference.py | 109 +++++++------------ app/main.py | 29 +++-- app/post_process.py | 37 ++++--- app/search/search_backend.py | 205 +++++++++++++++++++++++------------ app/search/search_manage.py | 153 ++++++++++++++------------ 5 files changed, 303 insertions(+), 230 deletions(-) diff --git a/app/inference.py b/app/inference.py index 0a38fe371a4..5336509dcf0 100644 --- a/app/inference.py +++ b/app/inference.py @@ -141,17 +141,7 @@ def run_one_task(task: Task, output_dir: str, model_names: Iterable[str]) -> boo return True -def select_patch(task: Task, output_dir: str | PathLike) -> tuple[str, dict]: - - patches = natsorted(list(Path(output_dir).glob("**/extracted_patch_*.diff"))) - - # TODO: These candidate patches must have been dismissed by reviewer. Maybe an - # assertion should be added to confirm this. - candidate_patches = [p for p in patches if may_pass_regression_tests(task, p)] - - agent_comment = None - thread = None - +def _find_reviewer_approved_patch(patches: list, candidate_patches: list): for p in candidate_patches: index = p.with_suffix("").name.rpartition("_")[2] reviews = natsorted( @@ -165,64 +155,49 @@ def select_patch(task: Task, output_dir: str | PathLike) -> tuple[str, dict]: assert samefile( p, last_patch ), f"{p} is approved and passes validation, but the last patch was {last_patch}" - selected_patch = p - reason = "reviewer-approved" - break - else: - if len(candidate_patches) > 1: - content_to_indices = defaultdict(list) - for idx, p in enumerate(candidate_patches): - content_to_indices[p.read_text()].append(idx) - items = sorted( - content_to_indices.items(), - key=lambda item: (len(item[1]), -item[1][0]), - reverse=True, - ) + return p + return None - # if len(items[0]) > 1: - if False: - index = items[0][1][0] - selected_patch = candidate_patches[index] - reason = "majority,multiple-pass-regression" - else: - try: - index, agent_comment, thread = agent_select.run( - task.get_issue_statement(), - [p.read_text() for p in candidate_patches], - ) - reason = "agent-selected,multiple-pass-regression" - except Exception: - index = -1 - reason = "agent-error,multiple-pass-regression" - selected_patch = candidate_patches[index] - elif len(candidate_patches) == 1: - selected_patch = candidate_patches[0] - reason = "no-agent,single-pass-regression" - else: - content_to_indices = defaultdict(list) - for idx, p in enumerate(patches): - content_to_indices[p.read_text()].append(idx) - items = sorted( - content_to_indices.items(), - key=lambda item: (len(item[1]), -item[1][0]), - reverse=True, - ) - # if len(items[0]) > 1: - if False: - index = items[0][1][0] - selected_patch = patches[index] - reason = "majority,none-pass-regression" - else: - try: - index, agent_comment, thread = agent_select.run( - task.get_issue_statement(), [p.read_text() for p in patches] - ) - reason = "agent-selected,none-pass-regression" - except Exception: - index = -1 - reason = "agent-error,none-pass-regression" - selected_patch = patches[index] +def _select_among_patches(task: Task, patches: list, reason_suffix: str): + agent_comment = None + thread = None + try: + index, agent_comment, thread = agent_select.run( + task.get_issue_statement(), [p.read_text() for p in patches] + ) + reason = f"agent-selected,{reason_suffix}" + except Exception: + index = -1 + reason = f"agent-error,{reason_suffix}" + return patches[index], reason, agent_comment, thread + + +def select_patch(task: Task, output_dir: str | PathLike) -> tuple[str, dict]: + + patches = natsorted(list(Path(output_dir).glob("**/extracted_patch_*.diff"))) + + # TODO: These candidate patches must have been dismissed by reviewer. Maybe an + # assertion should be added to confirm this. + candidate_patches = [p for p in patches if may_pass_regression_tests(task, p)] + + agent_comment = None + thread = None + + selected_patch = _find_reviewer_approved_patch(patches, candidate_patches) + if selected_patch is not None: + reason = "reviewer-approved" + elif len(candidate_patches) > 1: + selected_patch, reason, agent_comment, thread = _select_among_patches( + task, candidate_patches, "multiple-pass-regression" + ) + elif len(candidate_patches) == 1: + selected_patch = candidate_patches[0] + reason = "no-agent,single-pass-regression" + else: + selected_patch, reason, agent_comment, thread = _select_among_patches( + task, patches, "none-pass-regression" + ) rel_selected_patch = str(selected_patch.relative_to(output_dir)) diff --git a/app/main.py b/app/main.py index 85aed857203..ffa99cb073b 100644 --- a/app/main.py +++ b/app/main.py @@ -556,6 +556,24 @@ def evaluate_swe_issue_reproducer( task.evaluate_reproducer(reproducer_file, developer_patch_file, report_dir) +def _remove_task_output_files(task_output_dir: str, log_file_name: str) -> None: + """ + Remove everything in the task output dir other than the log file and + some meta data files dumped by RawTask. + """ + files_to_keep = [ + log_file_name, + "meta.json", + "problem_statement.txt", + "developer_patch.diff", + ] + for f in Path(task_output_dir).iterdir(): + if f.is_file() and f.name not in files_to_keep: + f.unlink() + if f.is_dir(): + shutil.rmtree(str(f)) + + def do_inference(python_task: Task, task_output_dir: str) -> bool: apputils.create_dir_if_not_exists(task_output_dir) @@ -605,16 +623,7 @@ def do_inference(python_task: Task, task_output_dir: str) -> bool: "Removing all files except info.log and meta files." ) - for f in Path(task_output_dir).iterdir(): - if f.is_file() and f.name not in [ - log_file_name, - "meta.json", - "problem_statement.txt", - "developer_patch.diff", - ]: - f.unlink() - if f.is_dir(): - shutil.rmtree(str(f)) + _remove_task_output_files(task_output_dir, log_file_name) run_ok = inference.run_one_task( python_task, task_output_dir, config.backup_model diff --git a/app/post_process.py b/app/post_process.py index edfc9a1e8d0..fa4b96898ac 100644 --- a/app/post_process.py +++ b/app/post_process.py @@ -215,6 +215,27 @@ def extract_diff_one_instance( return status, summary +def _apply_edits(edits, repo_path: str) -> list[int]: + """ + Try to match and apply each edit. Returns indexes of edits that could not be applied. + """ + unmatched_edit_indexes = [] + for idx, edit in enumerate(edits): + # NOTE: do not clean here, since we want to accumulate changes from all edits + target_file = edit.filename + # find the target file. The model may only use the short name of the file, + # so we need to search for it here + found_file = apputils.find_file(repo_path, target_file) + if found_file is None: + unmatched_edit_indexes.append(idx) + continue + # try to apply this edit and update the actual file content + applied_file = apply_edit(edit, found_file) + if applied_file is None: + unmatched_edit_indexes.append(idx) + return unmatched_edit_indexes + + # TODO: move this to PatchWriter def convert_response_to_diff( response: str, task_dir: str, standalone_mode: bool = False @@ -267,21 +288,7 @@ def convert_response_to_diff( # containing the test_patch content. We should just clean the changes until HEAD. apputils.repo_clean_changes() # try to match and apply each edit - unmatched_edit_indexes = [] - for idx, edit in enumerate(edits): - # NOTE: do not clean here, since we want to accumulate changes from all edits - target_file = edit.filename - # find the target file. The model may only use the short name of the file, - # so we need to search for it here - found_file = apputils.find_file(repo_path, target_file) - if found_file is None: - unmatched_edit_indexes.append(idx) - continue - # try to apply this edit and update the actual file content - applied_file = apply_edit(edit, found_file) - if applied_file is None: - unmatched_edit_indexes.append(idx) - continue + unmatched_edit_indexes = _apply_edits(edits, repo_path) if len(unmatched_edit_indexes) == len(edits): # non of the edits can be matched diff --git a/app/search/search_backend.py b/app/search/search_backend.py index 24749ef3f27..abba19546a4 100644 --- a/app/search/search_backend.py +++ b/app/search/search_backend.py @@ -770,17 +770,9 @@ def get_bug_loc_snippets_new(self, bug_location_dict: dict[str, str]): # (1) sometimes model can write class_name and method_name together in the # format Class.method - if not tmp_class_name and tmp_method_name and "." in tmp_method_name: - fragments = tmp_method_name.split(".") - if len(fragments) == 2: - tmp_class_name, tmp_method_name = fragments - logger.warning( - "Successfully split {} and {}", tmp_class_name, tmp_method_name - ) - else: - logger.warning( - "Too many fragments. Examine the method name: {}", tmp_method_name - ) + tmp_class_name, tmp_method_name = self._split_class_method_name( + tmp_class_name, tmp_method_name + ) # we require at least the file_name to be given assert ( @@ -802,83 +794,154 @@ def get_bug_loc_snippets_new(self, bug_location_dict: dict[str, str]): # satisfy the requirement above, so DO NOT use those functions here. if tmp_method_name and tmp_class_name: - output, curr_search_res, call_ok = self.search_method_in_class( - tmp_method_name, tmp_class_name + search_res, class_context_search_res, call_ok = ( + self._search_method_in_class_with_context( + tmp_method_name, tmp_class_name + ) ) - search_res.extend(curr_search_res) - if call_ok: - # when the location is decided to be method in a class, we also - # obtain (1) the entire class as location, and (2) the inherited - # parent class methods as location. - res: SearchResult - for res in curr_search_res: - - if ( - res.class_name is None - or res.func_name is None - or res.file_path is None - ): - continue - - inherited_output, inherited_search_res, _ = ( - self._get_inherited_methods(res.class_name, res.func_name) - ) - search_res.extend(inherited_search_res) - - # this kind of class is special, they just serve as a context, - # so the 'intended_behavior' field content does not apply for them. - - class_output, class_search_res, _ = self.search_class_in_file( - res.class_name, res.file_path - ) - class_context_search_res.extend(class_search_res) - - if (not call_ok) and tmp_method_name and tmp_file_name: - output, search_res, call_ok = self.search_method_in_file( - tmp_method_name, tmp_file_name + if not call_ok: + search_res, call_ok = self._search_fallback_locations( + tmp_file_name, tmp_method_name, tmp_class_name ) - if (not call_ok) and tmp_class_name and tmp_file_name: - output, search_res, call_ok = self.search_class_in_file( - tmp_class_name, tmp_file_name + if not call_ok: + # cannot find any location!! + return [] + + # we have some SearchResults => turn these into BugLocations + final_bug_locs = self._search_results_to_bug_locs( + search_res, intended_behavior + ) + + # deal with additional class context search results + final_bug_locs.extend( + self._search_results_to_bug_locs( + class_context_search_res, + "This class provides additional context to the issue.", ) + ) + + return final_bug_locs + + @staticmethod + def _split_class_method_name( + tmp_class_name: str, tmp_method_name: str + ) -> tuple[str, str]: + """ + Sometimes model can write class_name and method_name together in the + format Class.method. Split them if possible. + """ + if tmp_class_name or not tmp_method_name or "." not in tmp_method_name: + return tmp_class_name, tmp_method_name - if (not call_ok) and tmp_class_name: - output, search_res, call_ok = self.get_class_full_snippet(tmp_class_name) + fragments = tmp_method_name.split(".") + if len(fragments) != 2: + logger.warning( + "Too many fragments. Examine the method name: {}", tmp_method_name + ) + return tmp_class_name, tmp_method_name - if (not call_ok) and tmp_method_name: - output, search_res, call_ok = self.search_method(tmp_method_name) + tmp_class_name, tmp_method_name = fragments + logger.warning("Successfully split {} and {}", tmp_class_name, tmp_method_name) + return tmp_class_name, tmp_method_name - if (not call_ok) and tmp_file_name: - output, search_res, call_ok = self.get_file_content(tmp_file_name) + def _search_method_in_class_with_context( + self, method_name: str, class_name: str + ) -> tuple[list[SearchResult], list[SearchResult], bool]: + """ + Search for a method in a class. When found, also collect the inherited + parent class methods, and the entire class as context. + Returns (search_res, class_context_search_res, call_ok). + """ + class_context_search_res: list[SearchResult] = [] + _, curr_search_res, call_ok = self.search_method_in_class( + method_name, class_name + ) + search_res: list[SearchResult] = list(curr_search_res) if not call_ok: - # cannot find any location!! - return [] + return search_res, class_context_search_res, call_ok - # we have some SearchResults => turn these into BugLocations + # when the location is decided to be method in a class, we also + # obtain (1) the entire class as location, and (2) the inherited + # parent class methods as location. res: SearchResult + for res in curr_search_res: + if not self._has_class_func_and_file(res): + continue - final_bug_locs: list[BugLocation] = [] - for res in search_res: - if res.start is None or res.end is None: + _, inherited_search_res, _ = self._get_inherited_methods( + res.class_name, res.func_name + ) + search_res.extend(inherited_search_res) + + # this kind of class is special, they just serve as a context, + # so the 'intended_behavior' field content does not apply for them. + _, class_search_res, _ = self.search_class_in_file( + res.class_name, res.file_path + ) + class_context_search_res.extend(class_search_res) + + return search_res, class_context_search_res, call_ok + + @staticmethod + def _has_class_func_and_file(res: SearchResult) -> bool: + return ( + res.class_name is not None + and res.func_name is not None + and res.file_path is not None + ) + + def _search_fallback_locations( + self, tmp_file_name: str, tmp_method_name: str, tmp_class_name: str + ) -> tuple[list[SearchResult], bool]: + """ + Try progressively less specific searches until one succeeds. + Returns (search_res, call_ok). + """ + attempts = [ + ( + bool(tmp_method_name and tmp_file_name), + lambda: self.search_method_in_file(tmp_method_name, tmp_file_name), + ), + ( + bool(tmp_class_name and tmp_file_name), + lambda: self.search_class_in_file(tmp_class_name, tmp_file_name), + ), + ( + bool(tmp_class_name), + lambda: self.get_class_full_snippet(tmp_class_name), + ), + ( + bool(tmp_method_name), + lambda: self.search_method(tmp_method_name), + ), + ( + bool(tmp_file_name), + lambda: self.get_file_content(tmp_file_name), + ), + ] + + search_res: list[SearchResult] = [] + for should_try, search_func in attempts: + if not should_try: continue - new_bug_loc = BugLocation(res, self.project_path, intended_behavior) - final_bug_locs.append(new_bug_loc) + _, search_res, call_ok = search_func() + if call_ok: + return search_res, True - # deal with additional class context search results - for res in class_context_search_res: + return search_res, False + + def _search_results_to_bug_locs( + self, search_res: list[SearchResult], intended_behavior: str + ) -> list[BugLocation]: + bug_locs: list[BugLocation] = [] + for res in search_res: if res.start is None or res.end is None: continue - new_bug_loc = BugLocation( - res, - self.project_path, - "This class provides additional context to the issue.", - ) - final_bug_locs.append(new_bug_loc) - - return final_bug_locs + bug_locs.append(BugLocation(res, self.project_path, intended_behavior)) + return bug_locs if __name__ == "__main__": diff --git a/app/search/search_manage.py b/app/search/search_manage.py index e418333e877..7aa4bdda0d8 100644 --- a/app/search/search_manage.py +++ b/app/search/search_manage.py @@ -95,55 +95,14 @@ def search_iterative( json_api_calls = selected_apis_json.get("API_calls", []) buggy_locations = selected_apis_json.get("bug_locations", []) - formatted = [] - if json_api_calls: - formatted.append("API calls:") - for call in json_api_calls: - formatted.extend([f"\n- `{call}`"]) - - if buggy_locations: - formatted.append("\n\nBug locations") - for location in buggy_locations: - s = ", ".join(f"{k}: `{v}`" for k, v in location.items()) - formatted.extend([f"\n- {s}"]) - - print_acr("\n".join(formatted), "Agent-selected API calls") + self._print_selected_apis(json_api_calls, buggy_locations) # locations are confirmed by the agent - let's see whether the bug # locations are valid/precise if buggy_locations and (not json_api_calls): - # dump the locations for debugging - bug_loc_file = Path( - self.output_dir, "bug_locations_before_process.json" - ) - bug_loc_file.write_text(json.dumps(buggy_locations, indent=4)) - - new_bug_locations: list[BugLocation] = list() - - for loc in buggy_locations: - # this is the transformed bug location - new_bug_locations.extend(self.backend.get_bug_loc_snippets_new(loc)) - - # remove duplicates in the bug locations - unique_bug_locations: list[BugLocation] = [] - for loc in new_bug_locations: - if loc not in unique_bug_locations: - unique_bug_locations.append(loc) + new_bug_locations = self._process_bug_locations(buggy_locations) if new_bug_locations: - - # some locations can be extracted, good to proceed to patch gen - bug_loc_file_processed = Path( - self.output_dir, "bug_locations_after_process.json" - ) - - json_obj = [loc.to_dict() for loc in new_bug_locations] - bug_loc_file_processed.write_text(json.dumps(json_obj, indent=4)) - - logger.debug( - f"Bug location extracted successfully: {new_bug_locations}" - ) - return new_bug_locations, search_msg_thread # bug location is not precise enough to go into patch gen @@ -157,30 +116,7 @@ def search_iterative( continue # location not confirmed by the search agent - send backend result and go to next round - collated_search_res_str = "" - - for api_call in json_api_calls: - func_name, func_args = parse_function_invocation(api_call) - # TODO: there are currently duplicated code here and in agent_proxy. - func_unwrapped = getattr(self.backend, func_name) - while "__wrapped__" in func_unwrapped.__dict__: - func_unwrapped = func_unwrapped.__wrapped__ - arg_spec = inspect.getfullargspec(func_unwrapped) - arg_names = arg_spec.args[1:] # first parameter is self - - assert len(func_args) == len( - arg_names - ), f"Number of argument is wrong in API call: {api_call}" - - kwargs = dict(zip(arg_names, func_args)) - - function = getattr(self.backend, func_name) - result_str, _, call_ok = function(**kwargs) - collated_search_res_str += f"Result of {api_call}:\n\n" - collated_search_res_str += result_str + "\n\n" - - # record the api calls made and the call status - self.add_tool_call_to_curr_layer(func_name, kwargs, call_ok) + collated_search_res_str = self._invoke_api_calls(json_api_calls) print_acr(collated_search_res_str, f"context retrieval round {round_no}") # send the results back to the search agent @@ -195,6 +131,89 @@ def search_iterative( assert search_msg_thread is not None return [], search_msg_thread + @staticmethod + def _print_selected_apis(json_api_calls: list, buggy_locations: list) -> None: + formatted = [] + if json_api_calls: + formatted.append("API calls:") + for call in json_api_calls: + formatted.extend([f"\n- `{call}`"]) + + if buggy_locations: + formatted.append("\n\nBug locations") + for location in buggy_locations: + s = ", ".join(f"{k}: `{v}`" for k, v in location.items()) + formatted.extend([f"\n- {s}"]) + + print_acr("\n".join(formatted), "Agent-selected API calls") + + def _process_bug_locations(self, buggy_locations: list) -> list[BugLocation]: + """ + Transform agent-selected bug locations into BugLocation objects. + Returns an empty list if no location could be extracted. + """ + # dump the locations for debugging + bug_loc_file = Path(self.output_dir, "bug_locations_before_process.json") + bug_loc_file.write_text(json.dumps(buggy_locations, indent=4)) + + new_bug_locations: list[BugLocation] = list() + + for loc in buggy_locations: + # this is the transformed bug location + new_bug_locations.extend(self.backend.get_bug_loc_snippets_new(loc)) + + # remove duplicates in the bug locations + unique_bug_locations: list[BugLocation] = [] + for loc in new_bug_locations: + if loc not in unique_bug_locations: + unique_bug_locations.append(loc) + + if not new_bug_locations: + return new_bug_locations + + # some locations can be extracted, good to proceed to patch gen + bug_loc_file_processed = Path( + self.output_dir, "bug_locations_after_process.json" + ) + + json_obj = [loc.to_dict() for loc in new_bug_locations] + bug_loc_file_processed.write_text(json.dumps(json_obj, indent=4)) + + logger.debug(f"Bug location extracted successfully: {new_bug_locations}") + + return new_bug_locations + + def _invoke_api_calls(self, json_api_calls: list) -> str: + """ + Invoke the search API calls on the backend and collate their results. + """ + collated_search_res_str = "" + + for api_call in json_api_calls: + func_name, func_args = parse_function_invocation(api_call) + # TODO: there are currently duplicated code here and in agent_proxy. + func_unwrapped = getattr(self.backend, func_name) + while "__wrapped__" in func_unwrapped.__dict__: + func_unwrapped = func_unwrapped.__wrapped__ + arg_spec = inspect.getfullargspec(func_unwrapped) + arg_names = arg_spec.args[1:] # first parameter is self + + assert len(func_args) == len( + arg_names + ), f"Number of argument is wrong in API call: {api_call}" + + kwargs = dict(zip(arg_names, func_args)) + + function = getattr(self.backend, func_name) + result_str, _, call_ok = function(**kwargs) + collated_search_res_str += f"Result of {api_call}:\n\n" + collated_search_res_str += result_str + "\n\n" + + # record the api calls made and the call status + self.add_tool_call_to_curr_layer(func_name, kwargs, call_ok) + + return collated_search_res_str + def start_new_tool_call_layer(self): self.tool_call_layers.append([])