fix(run): use Fix: threading.local() in SaveApplyPatchHook to prevent race condition in multi-worker batch runs (fixes #1284) (#1366)

Root cause: on_instance_start stored env and problem_statement as plain instance attributes. With ThreadPoolExecutor, a shared hook instance was written by worker B before worker A's on_instance_completed ran, causing the wrong instance_id to be used and patches saved to the wrong directory.

Fix: store per-instance state in threading.local() so each worker thread gets an isolated copy.

sweagent/run/hooks/apply_patch.py

@@ -1,4 +1,5 @@
 import subprocess
+import threading
 from pathlib import Path
 
 import rich
@@ -21,27 +22,30 @@ class SaveApplyPatchHook(RunHook):
         self.logger = get_logger("swea-save_apply_patch", emoji="⚡️")
         self._apply_patch_locally = apply_patch_locally
         self._show_success_message = show_success_message
+        # Thread-local storage so that concurrent workers in run-batch do not
+        # overwrite each other's per-instance state (_env, _problem_statement).
+        self._local = threading.local()
 
     def on_init(self, *, run):
         self._output_dir = Path(run.output_dir)
 
     def on_instance_start(self, *, index: int, env: SWEEnv, problem_statement: ProblemStatementConfig):
-        self._env = env
-        self._problem_statement = problem_statement
+        self._local.env = env
+        self._local.problem_statement = problem_statement
 
     def on_instance_completed(self, *, result: AgentRunResult):
-        instance_id = self._problem_statement.id
+        instance_id = self._local.problem_statement.id
         patch_path = self._save_patch(instance_id, result.info)
         if patch_path:
             if not self._apply_patch_locally:
                 return
             if not _is_promising_patch(result.info):
                 return
-            if self._env.repo is None:
+            if self._local.env.repo is None:
                 return
-            if not isinstance(self._env.repo, LocalRepoConfig):
+            if not isinstance(self._local.env.repo, LocalRepoConfig):
                 return
-            local_dir = Path(self._env.repo.path)
+            local_dir = Path(self._local.env.repo.path)
             self._apply_patch(patch_path, local_dir)
 
     @staticmethod

tests/test_run_hooks.py

@@ -1,8 +1,12 @@
 import os
+import threading
+from concurrent.futures import ThreadPoolExecutor
+from unittest.mock import MagicMock
 
 import pytest
 
-from sweagent.agent.problem_statement import GithubIssue
+from sweagent.agent.problem_statement import GithubIssue, TextProblemStatement
+from sweagent.run.hooks.apply_patch import SaveApplyPatchHook
 from sweagent.run.hooks.open_pr import OpenPRConfig, OpenPRHook
 from sweagent.types import AgentRunResult
 
@@ -73,3 +77,53 @@ def test_should_open_pr_success_has_pr_override(open_pr_hook_init_for_sop, agent
     hook._problem_statement = GithubIssue(github_url="https://github.com/swe-agent/test-repo/issues/19")
     hook._config.skip_if_commits_reference_issue = False
     assert hook.should_open_pr(agent_run_result)
+
+
+def test_save_apply_patch_hook_concurrent_workers_save_to_correct_dirs(tmp_path):
+    """Regression test for #1284: concurrent workers must not overwrite each
+    other's per-instance state (_problem_statement, _env) in SaveApplyPatchHook.
+
+    Before the fix, a single shared hook instance stored _problem_statement as a
+    plain instance attribute.  When two workers both called on_instance_start()
+    before either finished, the second write overwrote the first, causing both
+    workers to save their patch into the same (wrong) output directory.
+
+    The fix uses threading.local() so every worker thread sees its own copy.
+    """
+    hook = SaveApplyPatchHook(show_success_message=False)
+    hook._output_dir = tmp_path
+
+    # Barrier ensures both workers have called on_instance_start before either
+    # proceeds to on_instance_completed, making the race window deterministic.
+    barrier = threading.Barrier(2)
+
+    def worker(instance_id: str, patch_content: str) -> None:
+        ps = TextProblemStatement(text=f"Issue for {instance_id}", id=instance_id)
+        env = MagicMock()
+        env.repo = None  # skip apply_patch_locally path
+
+        hook.on_instance_start(index=0, env=env, problem_statement=ps)
+
+        # Hold here until both threads have written their problem_statement so
+        # that the race window is guaranteed to be open.
+        barrier.wait()
+
+        result = AgentRunResult(
+            info={"submission": patch_content, "exit_status": "submitted"},
+            trajectory=[],
+        )
+        hook.on_instance_completed(result=result)
+
+    with ThreadPoolExecutor(max_workers=2) as pool:
+        fa = pool.submit(worker, "instance-A", "patch A content")
+        fb = pool.submit(worker, "instance-B", "patch B content")
+        fa.result()
+        fb.result()
+
+    patch_a = tmp_path / "instance-A" / "instance-A.patch"
+    patch_b = tmp_path / "instance-B" / "instance-B.patch"
+
+    assert patch_a.exists(), "Patch for instance-A was not saved to its own directory"
+    assert patch_b.exists(), "Patch for instance-B was not saved to its own directory"
+    assert patch_a.read_text() == "patch A content"
+    assert patch_b.read_text() == "patch B content"