Merge pull request #2 from Agent-Field/fix/concurrent-build-isolation

fix: isolate concurrent builds via build_id namespace on git resources

Author: AbirAbbas

swe_af/app.py

@@ -12,6 +12,7 @@
 import os
 import re
 import subprocess
+import uuid
 
 from swe_af.reasoners import router
 from swe_af.reasoners.pipeline import _assign_sequence_numbers, _compute_levels, _validate_file_conflicts
@@ -76,7 +77,8 @@ async def build(
         raise ValueError("Either repo_path or repo_url must be provided")
 
     # Clone if repo_url is set and target doesn't exist yet
-    if cfg.repo_url and not os.path.exists(os.path.join(repo_path, ".git")):
+    git_dir = os.path.join(repo_path, ".git")
+    if cfg.repo_url and not os.path.exists(git_dir):
         app.note(f"Cloning {cfg.repo_url} → {repo_path}", tags=["build", "clone"])
         os.makedirs(repo_path, exist_ok=True)
         clone_result = subprocess.run(
@@ -88,6 +90,58 @@ async def build(
             err = clone_result.stderr.strip()
             app.note(f"Clone failed (exit {clone_result.returncode}): {err}", tags=["build", "clone", "error"])
             raise RuntimeError(f"git clone failed (exit {clone_result.returncode}): {err}")
+    elif cfg.repo_url and os.path.exists(git_dir):
+        # Repo already cloned by a prior build — reset to remote default branch
+        # so git_init creates the integration branch from a clean baseline.
+        default_branch = cfg.github_pr_base or "main"
+        app.note(
+            f"Repo already exists at {repo_path} — resetting to origin/{default_branch}",
+            tags=["build", "clone", "reset"],
+        )
+
+        # Remove stale worktrees on disk before touching branches
+        worktrees_dir = os.path.join(repo_path, ".worktrees")
+        if os.path.isdir(worktrees_dir):
+            import shutil
+            shutil.rmtree(worktrees_dir, ignore_errors=True)
+        subprocess.run(
+            ["git", "worktree", "prune"],
+            cwd=repo_path, capture_output=True, text=True,
+        )
+
+        # Fetch latest remote state
+        fetch = subprocess.run(
+            ["git", "fetch", "origin"],
+            cwd=repo_path, capture_output=True, text=True,
+        )
+        if fetch.returncode != 0:
+            app.note(f"git fetch failed: {fetch.stderr.strip()}", tags=["build", "clone", "error"])
+
+        # Force-checkout default branch (handles dirty working tree from crashed builds)
+        subprocess.run(
+            ["git", "checkout", "-f", default_branch],
+            cwd=repo_path, capture_output=True, text=True,
+        )
+        reset = subprocess.run(
+            ["git", "reset", "--hard", f"origin/{default_branch}"],
+            cwd=repo_path, capture_output=True, text=True,
+        )
+        if reset.returncode != 0:
+            # Hard reset failed — nuke and re-clone as last resort
+            app.note(
+                f"Reset to origin/{default_branch} failed — re-cloning",
+                tags=["build", "clone", "reclone"],
+            )
+            import shutil
+            shutil.rmtree(repo_path, ignore_errors=True)
+            os.makedirs(repo_path, exist_ok=True)
+            clone_result = subprocess.run(
+                ["git", "clone", cfg.repo_url, repo_path],
+                capture_output=True, text=True,
+            )
+            if clone_result.returncode != 0:
+                err = clone_result.stderr.strip()
+                raise RuntimeError(f"git re-clone failed: {err}")
     else:
         # Ensure repo_path exists even when no repo_url is provided (fresh init case)
         # This is needed because planning agents may need to read the repo in parallel with git_init
@@ -105,7 +159,11 @@ async def build(
     # Resolve runtime + flat model config once for this build.
     resolved = cfg.resolved_models()
 
-    app.note("Build starting", tags=["build", "start"])
+    # Unique ID for this build — namespaces git branches/worktrees to prevent
+    # collisions when multiple builds run concurrently on the same repository.
+    build_id = uuid.uuid4().hex[:8]
+
+    app.note(f"Build starting (build_id={build_id})", tags=["build", "start"])
 
     # Compute absolute artifacts directory path for logging
     abs_artifacts_dir = os.path.join(os.path.abspath(repo_path), artifacts_dir)
@@ -151,6 +209,7 @@ async def build(
             permission_mode=cfg.permission_mode,
             ai_provider=cfg.ai_provider,
             previous_error=previous_error,
+            build_id=build_id,
         )
 
         # Run planning only on first attempt, then just git_init on retries
@@ -223,6 +282,7 @@ async def build(
         execute_fn_target=cfg.execute_fn_target,
         config=exec_config,
         git_config=git_config,
+        build_id=build_id,
     ), "execute")
 
     # 3. VERIFY
@@ -634,6 +694,7 @@ async def execute(
     config: dict | None = None,
     git_config: dict | None = None,
     resume: bool = False,
+    build_id: str = "",
 ) -> dict:
     """Execute a planned DAG with self-healing replanning.
 
@@ -675,6 +736,7 @@ async def execute_fn(issue, dag_state):
         node_id=NODE_ID,
         git_config=git_config,
         resume=resume,
+        build_id=build_id,
     )
     return state.model_dump()
 

swe_af/execution/dag_executor.py

@@ -56,6 +56,7 @@ async def _setup_worktrees(
     node_id: str,
     config: ExecutionConfig,
     note_fn: Callable | None = None,
+    build_id: str = "",
 ) -> list[dict]:
     """Create git worktrees for parallel issue isolation.
 
@@ -78,6 +79,7 @@ async def _setup_worktrees(
         level=dag_state.current_level,
         model=config.git_model,
         ai_provider=config.ai_provider,
+        build_id=build_id,
     )
 
     if not setup.get("success"):
@@ -365,7 +367,7 @@ def _load_checkpoint(artifacts_dir: str) -> DAGState | None:
 
 
 def _init_dag_state(
-    plan_result: dict, repo_path: str, git_config: dict | None = None,
+    plan_result: dict, repo_path: str, git_config: dict | None = None, build_id: str = "",
 ) -> DAGState:
     """Extract DAGState from a PlanResult dict.
 
@@ -424,6 +426,7 @@ def _init_dag_state(
         architecture_summary=architecture_summary,
         all_issues=all_issues,
         levels=levels,
+        build_id=build_id,
         **git_kwargs,
     )
 
@@ -978,6 +981,7 @@ async def run_dag(
     node_id: str = "swe-planner",
     git_config: dict | None = None,
     resume: bool = False,
+    build_id: str = "",
 ) -> DAGState:
     """Execute a planned DAG with self-healing replanning.
 
@@ -1022,7 +1026,7 @@ async def call_fn(target: str, **kwargs):
             result = await _raw_call_fn(target, **kwargs)
             return unwrap_call_result(result, target)
 
-    dag_state = _init_dag_state(plan_result, repo_path, git_config=git_config)
+    dag_state = _init_dag_state(plan_result, repo_path, git_config=git_config, build_id=build_id)
     dag_state.max_replans = config.max_replans
 
     # Resume from checkpoint if requested
@@ -1094,6 +1098,7 @@ async def _memory_fn(action: str, key: str, value=None):
         if call_fn and dag_state.git_integration_branch:
             active_issues = await _setup_worktrees(
                 dag_state, active_issues, call_fn, node_id, config, note_fn,
+                build_id=dag_state.build_id,
             )
 
         # Track in-flight issues and checkpoint before execution (Bug 4 fix)
@@ -1145,8 +1150,16 @@ async def _memory_fn(action: str, key: str, value=None):
                 )
 
             # Start cleanup in background (doesn't affect replan decisions)
+            # Use branch_name if injected by _setup_worktrees (includes build_id prefix),
+            # otherwise derive from build_id + sequence + name.
+            _bid = dag_state.build_id
             branches_to_clean = [
-                f"issue/{str(i.get('sequence_number') or 0).zfill(2)}-{i['name']}" for i in active_issues
+                i["branch_name"] if i.get("branch_name") else (
+                    f"issue/{_bid}-{str(i.get('sequence_number') or 0).zfill(2)}-{i['name']}"
+                    if _bid else
+                    f"issue/{str(i.get('sequence_number') or 0).zfill(2)}-{i['name']}"
+                )
+                for i in active_issues
             ]
             cleanup_task = asyncio.create_task(
                 _cleanup_worktrees(
@@ -1323,8 +1336,12 @@ async def _memory_fn(action: str, key: str, value=None):
 
     # Final worktree sweep — catch anything the per-level cleanup missed
     if call_fn and dag_state.worktrees_dir and dag_state.git_integration_branch:
-        # Collect all issue branches that should have been cleaned
+        # Collect all issue branches that should have been cleaned.
+        # Must use the same build_id-prefixed format that workspace setup created.
+        _bid = dag_state.build_id
         all_branches = [
+            f"issue/{_bid}-{str(i.get('sequence_number') or 0).zfill(2)}-{i['name']}"
+            if _bid else
             f"issue/{str(i.get('sequence_number') or 0).zfill(2)}-{i['name']}"
             for i in dag_state.all_issues
         ]

swe_af/execution/schemas.py

@@ -183,6 +183,7 @@ class DAGState(BaseModel):
     merged_branches: list[str] = []
     unmerged_branches: list[str] = []  # branches that failed to merge
     worktrees_dir: str = ""  # e.g. repo_path/.worktrees
+    build_id: str = ""  # unique per build() call; namespaces git branches/worktrees
 
     # --- Merge/test history ---
     merge_results: list[dict] = []

swe_af/prompts/git_init.py

@@ -28,7 +28,9 @@
 
 1. Record the current branch as `original_branch`.
 2. Ensure the working tree is clean (warn if not, but proceed).
-3. Create an integration branch: `git checkout -b feature/<goal-slug>` from HEAD.
+3. Create an integration branch from HEAD:
+   - If a **Build ID** is provided in the task: `git checkout -b feature/<build-id>-<goal-slug>`
+   - Otherwise: `git checkout -b feature/<goal-slug>`
 4. Record the initial commit SHA (HEAD before any work).
 
 ## Worktrees Directory
@@ -82,13 +84,15 @@
 """
 
 
-def git_init_task_prompt(repo_path: str, goal: str) -> str:
+def git_init_task_prompt(repo_path: str, goal: str, build_id: str = "") -> str:
     """Build the task prompt for the git initialization agent."""
     sections: list[str] = []
 
     sections.append("## Repository Setup Task")
     sections.append(f"- **Repository path**: `{repo_path}`")
     sections.append(f"- **Project goal**: {goal}")
+    if build_id:
+        sections.append(f"- **Build ID**: `{build_id}` (prefix integration branch slug with this)")
 
     sections.append(
         "\n## Your Task\n"

swe_af/prompts/workspace.py

@@ -15,15 +15,22 @@
 
 ## Your Responsibilities
 
-For each issue in this level, create a worktree:
+For each issue in this level, create a worktree using the **exact command format specified in the task**.
+The task will provide either a plain format or a Build-ID-prefixed format — always follow the task.
 
+Default (no Build ID):
 ```bash
 git worktree add <worktrees_dir>/issue-<NN>-<name> -b issue/<NN>-<name> <integration_branch>
 ```
 
+With Build ID (when the task specifies one — CRITICAL: you MUST use this form):
+```bash
+git worktree add <worktrees_dir>/issue-<BUILD_ID>-<NN>-<name> -b issue/<BUILD_ID>-<NN>-<name> <integration_branch>
+```
+
 This creates:
-- A new directory at `<worktrees_dir>/issue-<NN>-<name>`
-- A new branch `issue/<NN>-<name>` starting from the integration branch
+- A new directory at the worktrees path
+- A new branch starting from the integration branch
 - An isolated working copy where the coder agent can freely edit files
 
 ## Output
@@ -34,7 +41,7 @@
 
 ## Constraints
 
-- If a branch `issue/<NN>-<name>` already exists, remove the old worktree first and recreate.
+- If a branch with the target name already exists, remove the old worktree first and recreate.
 - All worktree operations must be run from the main repository directory.
 - Do NOT modify any source files — only git worktree commands.
 
@@ -95,6 +102,7 @@ def workspace_setup_task_prompt(
     integration_branch: str,
     issues: list[dict],
     worktrees_dir: str,
+    build_id: str = "",
 ) -> str:
     """Build the task prompt for the workspace setup agent."""
     sections: list[str] = []
@@ -103,6 +111,8 @@ def workspace_setup_task_prompt(
     sections.append(f"- **Repository path**: `{repo_path}`")
     sections.append(f"- **Integration branch**: `{integration_branch}`")
     sections.append(f"- **Worktrees directory**: `{worktrees_dir}`")
+    if build_id:
+        sections.append(f"- **Build ID**: `{build_id}`")
 
     sections.append("\n### Issues to create worktrees for:")
     for issue in issues:
@@ -111,16 +121,38 @@ def workspace_setup_task_prompt(
         seq = str(issue.get("sequence_number") or 0).zfill(2)
         sections.append(f"- issue_name=`{name}`, seq=`{seq}`, title: {title}")
 
-    sections.append(
+    if build_id:
+        worktree_cmd = (
+            f"git worktree add <worktrees_dir>/issue-{build_id}-<NN>-<name>"
+            f" -b issue/{build_id}-<NN>-<name> <integration_branch>"
+        )
+        branch_note = (
+            f"   Branch names MUST be prefixed with the Build ID: `issue/{build_id}-<NN>-<name>`\n"
+            f"   Worktree dirs MUST be prefixed with the Build ID: `issue-{build_id}-<NN>-<name>`\n"
+            "   This prevents collisions with other concurrent builds on the same repository."
+        )
+    else:
+        worktree_cmd = (
+            "git worktree add <worktrees_dir>/issue-<NN>-<name>"
+            " -b issue/<NN>-<name> <integration_branch>"
+        )
+        branch_note = ""
+
+    task = (
         "\n## Your Task\n"
         "1. Ensure you are in the main repository directory.\n"
         "2. For each issue, create a worktree:\n"
-        "   `git worktree add <worktrees_dir>/issue-<NN>-<name> -b issue/<NN>-<name> <integration_branch>`\n"
+        f"   `{worktree_cmd}`\n"
+    )
+    if branch_note:
+        task += branch_note + "\n"
+    task += (
         "3. Verify each worktree was created successfully.\n"
         "4. Return a JSON object with `workspaces` and `success`.\n\n"
         "IMPORTANT: In the output JSON, `issue_name` must be the canonical name "
         "(e.g. `value-copy-trait`), NOT the sequence-prefixed name (e.g. `01-value-copy-trait`)."
     )
+    sections.append(task)
 
     return "\n".join(sections)
 

swe_af/reasoners/execution_agents.py

@@ -520,6 +520,7 @@ async def run_git_init(
     permission_mode: str = "",
     ai_provider: str = "claude",
     previous_error: str | None = None,
+    build_id: str = "",
 ) -> dict:
     """Initialize git repo and create integration branch for feature work.
 
@@ -538,7 +539,7 @@ async def run_git_init(
         tags=["git_init", "start"],
     )
 
-    task_prompt = git_init_task_prompt(repo_path=repo_path, goal=goal)
+    task_prompt = git_init_task_prompt(repo_path=repo_path, goal=goal, build_id=build_id)
 
     # Build system prompt with error context if retrying
     system_prompt = GIT_INIT_SYSTEM_PROMPT
@@ -604,6 +605,7 @@ async def run_workspace_setup(
     model: str = "sonnet",
     permission_mode: str = "",
     ai_provider: str = "claude",
+    build_id: str = "",
 ) -> dict:
     """Create git worktrees for parallel issue isolation.
 
@@ -623,6 +625,7 @@ async def run_workspace_setup(
         integration_branch=integration_branch,
         issues=issues,
         worktrees_dir=worktrees_dir,
+        build_id=build_id,
     )
 
     class WorkspaceSetupResult(BaseModel):