Reduce cognitive complexity in 5 core functions via helper extraction

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))
 

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

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