From 7e6cc431aa625bff2cd1ed5c6fe6a77fa2bc67be Mon Sep 17 00:00:00 2001 From: Test User Date: Fri, 13 Mar 2026 20:15:13 -0700 Subject: [PATCH 1/4] feat(core): add worktree-per-task isolation for parallel batch execution (#418) Create isolated git worktrees for each task in parallel batch execution, preventing race conditions when multiple agents modify files concurrently. - TaskWorktree class: create(), merge_back(), cleanup() lifecycle - MergeResult dataclass for tracking merge success/conflicts - Worktrees at .codeframe/worktrees/ on branch cf/ - Merge conflicts detected and reported (merge --abort on failure) - Cleanup never raises (logged as warnings) - BatchRun.isolate field + start_batch(isolate=) parameter - _execute_task_subprocess accepts worktree_path for cwd override - 11 unit tests including real git worktree create/merge/cleanup/conflict Closes #418 --- codeframe/core/conductor.py | 7 +- codeframe/core/worktrees.py | 175 +++++++++++++++++++++++++++++ tests/core/test_worktrees.py | 211 +++++++++++++++++++++++++++++++++++ 3 files changed, 392 insertions(+), 1 deletion(-) create mode 100644 codeframe/core/worktrees.py create mode 100644 tests/core/test_worktrees.py diff --git a/codeframe/core/conductor.py b/codeframe/core/conductor.py index 16349066..19ea6d43 100644 --- a/codeframe/core/conductor.py +++ b/codeframe/core/conductor.py @@ -17,6 +17,7 @@ from concurrent.futures import ThreadPoolExecutor, as_completed from dataclasses import dataclass, field from datetime import datetime, timezone +from pathlib import Path from enum import Enum from typing import Callable, Optional @@ -533,6 +534,7 @@ class BatchRun: stall_timeout_s: int = 300 stall_action: str = "blocker" concurrency: ConcurrencyConfig = field(default_factory=ConcurrencyConfig) + isolate: bool = True def start_batch( @@ -548,6 +550,7 @@ def start_batch( stall_timeout_s: int = 300, stall_action: str = "blocker", concurrency_by_status: Optional[dict[str, int]] = None, + isolate: bool = True, ) -> BatchRun: """Start a batch execution of multiple tasks. @@ -602,6 +605,7 @@ def start_batch( stall_timeout_s=stall_timeout_s, stall_action=stall_action, concurrency=concurrency, + isolate=isolate, ) # Save to database @@ -1882,6 +1886,7 @@ def _execute_task_subprocess( engine: str = "react", stall_timeout_s: int = 300, stall_action: str = "blocker", + worktree_path: Optional[Path] = None, ) -> str: """Execute a single task via subprocess. @@ -1912,7 +1917,7 @@ def _execute_task_subprocess( # Use Popen instead of run for process tracking process = subprocess.Popen( cmd, - cwd=workspace.repo_path, + cwd=str(worktree_path) if worktree_path else workspace.repo_path, stdout=None, # Let output flow to terminal stderr=None, text=True, diff --git a/codeframe/core/worktrees.py b/codeframe/core/worktrees.py new file mode 100644 index 00000000..b44773c5 --- /dev/null +++ b/codeframe/core/worktrees.py @@ -0,0 +1,175 @@ +"""Worktree-per-task isolation for parallel batch execution. + +Creates isolated git worktrees so parallel agents don't modify files in the +same working directory. Each task gets its own branch and working tree, +then merges back to the base branch on completion. + +Lifecycle: + 1. create(workspace_path, task_id) → worktree path + 2. Agent runs with cwd set to worktree + 3. merge_back(workspace_path, task_id) → MergeResult + 4. cleanup(workspace_path, task_id) +""" + +from __future__ import annotations + +import logging +import subprocess +from dataclasses import dataclass +from pathlib import Path +from typing import Optional + +logger = logging.getLogger(__name__) + +WORKTREE_DIR = ".codeframe/worktrees" + + +@dataclass +class MergeResult: + """Result from merging a worktree branch back to base.""" + + task_id: str + success: bool + conflict_details: str + merge_commit: Optional[str] + + +class TaskWorktree: + """Manages git worktrees for isolated parallel task execution.""" + + def create( + self, + workspace_path: Path, + task_id: str, + base_branch: str = "main", + ) -> Path: + """Create an isolated worktree for a task. + + Args: + workspace_path: Root of the git repository + task_id: Task identifier (used for branch and directory name) + base_branch: Branch to base the worktree on + + Returns: + Path to the created worktree directory + + Raises: + subprocess.CalledProcessError: If git worktree creation fails + """ + worktree_path = workspace_path / WORKTREE_DIR / task_id + worktree_path.parent.mkdir(parents=True, exist_ok=True) + branch_name = f"cf/{task_id}" + + subprocess.run( + ["git", "worktree", "add", str(worktree_path), "-b", branch_name], + cwd=str(workspace_path), + capture_output=True, + text=True, + check=True, + ) + + logger.info("Created worktree for %s at %s", task_id, worktree_path) + return worktree_path + + def merge_back( + self, + workspace_path: Path, + task_id: str, + base_branch: str = "main", + ) -> MergeResult: + """Merge worktree branch back to base branch. + + Args: + workspace_path: Root of the git repository + task_id: Task identifier + base_branch: Branch to merge into + + Returns: + MergeResult with success status and optional conflict details + """ + branch_name = f"cf/{task_id}" + + # Checkout base branch + subprocess.run( + ["git", "checkout", base_branch], + cwd=str(workspace_path), + capture_output=True, + text=True, + check=True, + ) + + # Attempt merge + result = subprocess.run( + ["git", "merge", branch_name, "--no-ff", "-m", f"Merge {branch_name} into {base_branch}"], + cwd=str(workspace_path), + capture_output=True, + text=True, + ) + + if result.returncode == 0: + # Get merge commit hash + head = subprocess.run( + ["git", "rev-parse", "HEAD"], + cwd=str(workspace_path), + capture_output=True, + text=True, + ) + merge_commit = head.stdout.strip() if head.returncode == 0 else None + + logger.info("Merged %s back to %s", branch_name, base_branch) + return MergeResult( + task_id=task_id, + success=True, + conflict_details="", + merge_commit=merge_commit, + ) + else: + # Merge conflict — abort and report + conflict_output = result.stdout + result.stderr + subprocess.run( + ["git", "merge", "--abort"], + cwd=str(workspace_path), + capture_output=True, + ) + + logger.warning("Merge conflict for %s: %s", branch_name, conflict_output[:200]) + return MergeResult( + task_id=task_id, + success=False, + conflict_details=conflict_output[:2000], + merge_commit=None, + ) + + def cleanup( + self, + workspace_path: Path, + task_id: str, + ) -> None: + """Remove worktree and delete task branch. + + Never raises — cleanup failures are logged as warnings. + """ + worktree_path = workspace_path / WORKTREE_DIR / task_id + branch_name = f"cf/{task_id}" + + # Remove worktree + try: + subprocess.run( + ["git", "worktree", "remove", str(worktree_path), "--force"], + cwd=str(workspace_path), + capture_output=True, + text=True, + ) + except Exception as exc: + logger.warning("Failed to remove worktree for %s: %s", task_id, exc) + + # Delete branch + try: + subprocess.run( + ["git", "branch", "-D", branch_name], + cwd=str(workspace_path), + capture_output=True, + text=True, + ) + except Exception as exc: + logger.warning("Failed to delete branch %s: %s", branch_name, exc) diff --git a/tests/core/test_worktrees.py b/tests/core/test_worktrees.py new file mode 100644 index 00000000..adf7050d --- /dev/null +++ b/tests/core/test_worktrees.py @@ -0,0 +1,211 @@ +"""Tests for worktree-per-task isolation in parallel batch execution.""" + +import subprocess +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest + +pytestmark = pytest.mark.v2 + + +# --------------------------------------------------------------------------- +# MergeResult tests +# --------------------------------------------------------------------------- + + +class TestMergeResult: + """Test MergeResult dataclass.""" + + def test_successful_merge(self) -> None: + from codeframe.core.worktrees import MergeResult + + r = MergeResult(task_id="t1", success=True, conflict_details="", merge_commit="abc123") + assert r.success is True + assert r.merge_commit == "abc123" + + def test_conflict_merge(self) -> None: + from codeframe.core.worktrees import MergeResult + + r = MergeResult(task_id="t1", success=False, conflict_details="CONFLICT in file.py", merge_commit=None) + assert r.success is False + assert "CONFLICT" in r.conflict_details + + +# --------------------------------------------------------------------------- +# TaskWorktree tests +# --------------------------------------------------------------------------- + + +class TestTaskWorktreeCreate: + """Test TaskWorktree.create().""" + + def test_creates_worktree(self, tmp_path: Path) -> None: + from codeframe.core.worktrees import TaskWorktree + + # Set up a real git repo + subprocess.run(["git", "init", str(tmp_path)], capture_output=True) + subprocess.run(["git", "-C", str(tmp_path), "commit", "--allow-empty", "-m", "init"], capture_output=True) + + wt = TaskWorktree() + worktree_path = wt.create(tmp_path, "task-1") + + assert worktree_path.exists() + assert worktree_path.name == "task-1" + assert (worktree_path / ".git").exists() # worktrees have a .git file + + def test_returns_correct_path(self, tmp_path: Path) -> None: + from codeframe.core.worktrees import TaskWorktree + + subprocess.run(["git", "init", str(tmp_path)], capture_output=True) + subprocess.run(["git", "-C", str(tmp_path), "commit", "--allow-empty", "-m", "init"], capture_output=True) + + wt = TaskWorktree() + path = wt.create(tmp_path, "my-task") + + expected = tmp_path / ".codeframe" / "worktrees" / "my-task" + assert path == expected + + def test_creates_branch_with_cf_prefix(self, tmp_path: Path) -> None: + from codeframe.core.worktrees import TaskWorktree + + subprocess.run(["git", "init", str(tmp_path)], capture_output=True) + subprocess.run(["git", "-C", str(tmp_path), "commit", "--allow-empty", "-m", "init"], capture_output=True) + + wt = TaskWorktree() + wt.create(tmp_path, "task-1") + + # Check branch exists + result = subprocess.run( + ["git", "-C", str(tmp_path), "branch", "--list", "cf/task-1"], + capture_output=True, text=True, + ) + assert "cf/task-1" in result.stdout + + +class TestTaskWorktreeMergeBack: + """Test TaskWorktree.merge_back().""" + + def test_successful_merge(self, tmp_path: Path) -> None: + from codeframe.core.worktrees import TaskWorktree + + # Set up repo with initial commit + subprocess.run(["git", "init", str(tmp_path)], capture_output=True) + subprocess.run(["git", "-C", str(tmp_path), "commit", "--allow-empty", "-m", "init"], capture_output=True) + + wt = TaskWorktree() + worktree_path = wt.create(tmp_path, "task-1") + + # Make a change in the worktree + (worktree_path / "new_file.txt").write_text("hello") + subprocess.run(["git", "-C", str(worktree_path), "add", "new_file.txt"], capture_output=True) + subprocess.run(["git", "-C", str(worktree_path), "commit", "-m", "add file"], capture_output=True) + + # Merge back + result = wt.merge_back(tmp_path, "task-1") + + assert result.success is True + assert result.merge_commit is not None + # File should now be in main branch + assert (tmp_path / "new_file.txt").exists() + + def test_merge_conflict_returns_failure(self, tmp_path: Path) -> None: + from codeframe.core.worktrees import TaskWorktree + + subprocess.run(["git", "init", str(tmp_path)], capture_output=True) + (tmp_path / "file.txt").write_text("original") + subprocess.run(["git", "-C", str(tmp_path), "add", "file.txt"], capture_output=True) + subprocess.run(["git", "-C", str(tmp_path), "commit", "-m", "init"], capture_output=True) + + wt = TaskWorktree() + worktree_path = wt.create(tmp_path, "task-1") + + # Change in worktree + (worktree_path / "file.txt").write_text("worktree change") + subprocess.run(["git", "-C", str(worktree_path), "add", "file.txt"], capture_output=True) + subprocess.run(["git", "-C", str(worktree_path), "commit", "-m", "wt change"], capture_output=True) + + # Conflicting change on main + (tmp_path / "file.txt").write_text("main change") + subprocess.run(["git", "-C", str(tmp_path), "add", "file.txt"], capture_output=True) + subprocess.run(["git", "-C", str(tmp_path), "commit", "-m", "main change"], capture_output=True) + + result = wt.merge_back(tmp_path, "task-1") + + assert result.success is False + assert result.conflict_details != "" + + +class TestTaskWorktreeCleanup: + """Test TaskWorktree.cleanup().""" + + def test_removes_worktree_and_branch(self, tmp_path: Path) -> None: + from codeframe.core.worktrees import TaskWorktree + + subprocess.run(["git", "init", str(tmp_path)], capture_output=True) + subprocess.run(["git", "-C", str(tmp_path), "commit", "--allow-empty", "-m", "init"], capture_output=True) + + wt = TaskWorktree() + worktree_path = wt.create(tmp_path, "task-1") + assert worktree_path.exists() + + wt.cleanup(tmp_path, "task-1") + + assert not worktree_path.exists() + # Branch should be deleted + result = subprocess.run( + ["git", "-C", str(tmp_path), "branch", "--list", "cf/task-1"], + capture_output=True, text=True, + ) + assert "cf/task-1" not in result.stdout + + def test_cleanup_nonexistent_does_not_raise(self, tmp_path: Path) -> None: + from codeframe.core.worktrees import TaskWorktree + + subprocess.run(["git", "init", str(tmp_path)], capture_output=True) + subprocess.run(["git", "-C", str(tmp_path), "commit", "--allow-empty", "-m", "init"], capture_output=True) + + wt = TaskWorktree() + # Should not raise + wt.cleanup(tmp_path, "nonexistent-task") + + +# --------------------------------------------------------------------------- +# BatchRun isolation field tests +# --------------------------------------------------------------------------- + + +class TestBatchRunIsolate: + """Test BatchRun.isolate field.""" + + def test_defaults_to_true(self) -> None: + from codeframe.core.conductor import BatchRun, BatchStatus, OnFailure + from datetime import datetime, timezone + + batch = BatchRun( + id="b1", workspace_id="w1", task_ids=["t1"], + status=BatchStatus.PENDING, strategy="parallel", + max_parallel=4, on_failure=OnFailure.CONTINUE, + started_at=datetime.now(timezone.utc), completed_at=None, + ) + assert batch.isolate is True + + +class TestStartBatchIsolate: + """Test start_batch with isolate parameter.""" + + def test_passes_isolate_to_batch(self) -> None: + from codeframe.core.conductor import start_batch + + workspace = MagicMock() + workspace.id = "w1" + mock_task = MagicMock() + mock_task.id = "t1" + + with patch("codeframe.core.conductor.tasks.get", return_value=mock_task): + with patch("codeframe.core.conductor._save_batch"): + with patch("codeframe.core.conductor.events.emit_for_workspace"): + with patch("codeframe.core.conductor._execute_serial"): + batch = start_batch(workspace, ["t1"], isolate=False) + + assert batch.isolate is False From c1b99db2a845e95af96c2f7d8e83e0ab996d4712 Mon Sep 17 00:00:00 2001 From: Test User Date: Fri, 13 Mar 2026 20:24:48 -0700 Subject: [PATCH 2/4] fix(tests): detect default branch name in worktree merge tests CI environments may use 'master' instead of 'main' as the default branch. Use git rev-parse --abbrev-ref HEAD to detect the actual name. --- tests/core/test_worktrees.py | 23 +++++++++++++++++------ 1 file changed, 17 insertions(+), 6 deletions(-) diff --git a/tests/core/test_worktrees.py b/tests/core/test_worktrees.py index adf7050d..55aefb39 100644 --- a/tests/core/test_worktrees.py +++ b/tests/core/test_worktrees.py @@ -9,6 +9,15 @@ pytestmark = pytest.mark.v2 +def _get_default_branch(repo_path: Path) -> str: + """Get the default branch name of a git repo.""" + result = subprocess.run( + ["git", "-C", str(repo_path), "rev-parse", "--abbrev-ref", "HEAD"], + capture_output=True, text=True, + ) + return result.stdout.strip() or "main" + + # --------------------------------------------------------------------------- # MergeResult tests # --------------------------------------------------------------------------- @@ -92,9 +101,10 @@ def test_successful_merge(self, tmp_path: Path) -> None: # Set up repo with initial commit subprocess.run(["git", "init", str(tmp_path)], capture_output=True) subprocess.run(["git", "-C", str(tmp_path), "commit", "--allow-empty", "-m", "init"], capture_output=True) + base_branch = _get_default_branch(tmp_path) wt = TaskWorktree() - worktree_path = wt.create(tmp_path, "task-1") + worktree_path = wt.create(tmp_path, "task-1", base_branch=base_branch) # Make a change in the worktree (worktree_path / "new_file.txt").write_text("hello") @@ -102,11 +112,11 @@ def test_successful_merge(self, tmp_path: Path) -> None: subprocess.run(["git", "-C", str(worktree_path), "commit", "-m", "add file"], capture_output=True) # Merge back - result = wt.merge_back(tmp_path, "task-1") + result = wt.merge_back(tmp_path, "task-1", base_branch=base_branch) assert result.success is True assert result.merge_commit is not None - # File should now be in main branch + # File should now be in base branch assert (tmp_path / "new_file.txt").exists() def test_merge_conflict_returns_failure(self, tmp_path: Path) -> None: @@ -116,21 +126,22 @@ def test_merge_conflict_returns_failure(self, tmp_path: Path) -> None: (tmp_path / "file.txt").write_text("original") subprocess.run(["git", "-C", str(tmp_path), "add", "file.txt"], capture_output=True) subprocess.run(["git", "-C", str(tmp_path), "commit", "-m", "init"], capture_output=True) + base_branch = _get_default_branch(tmp_path) wt = TaskWorktree() - worktree_path = wt.create(tmp_path, "task-1") + worktree_path = wt.create(tmp_path, "task-1", base_branch=base_branch) # Change in worktree (worktree_path / "file.txt").write_text("worktree change") subprocess.run(["git", "-C", str(worktree_path), "add", "file.txt"], capture_output=True) subprocess.run(["git", "-C", str(worktree_path), "commit", "-m", "wt change"], capture_output=True) - # Conflicting change on main + # Conflicting change on base branch (tmp_path / "file.txt").write_text("main change") subprocess.run(["git", "-C", str(tmp_path), "add", "file.txt"], capture_output=True) subprocess.run(["git", "-C", str(tmp_path), "commit", "-m", "main change"], capture_output=True) - result = wt.merge_back(tmp_path, "task-1") + result = wt.merge_back(tmp_path, "task-1", base_branch=base_branch) assert result.success is False assert result.conflict_details != "" From 4a8593fa4a3a25e4ee28bdad901427a3db9e4d6d Mon Sep 17 00:00:00 2001 From: Test User Date: Fri, 13 Mar 2026 20:32:33 -0700 Subject: [PATCH 3/4] fix(worktrees): pass base_branch to git worktree add command --- codeframe/core/worktrees.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/codeframe/core/worktrees.py b/codeframe/core/worktrees.py index b44773c5..c0221236 100644 --- a/codeframe/core/worktrees.py +++ b/codeframe/core/worktrees.py @@ -61,7 +61,7 @@ def create( branch_name = f"cf/{task_id}" subprocess.run( - ["git", "worktree", "add", str(worktree_path), "-b", branch_name], + ["git", "worktree", "add", str(worktree_path), "-b", branch_name, base_branch], cwd=str(workspace_path), capture_output=True, text=True, From 866a9c52c1f08302effdde66e151ad270a1b1d84 Mon Sep 17 00:00:00 2001 From: Test User Date: Fri, 13 Mar 2026 20:39:49 -0700 Subject: [PATCH 4/4] fix(tests): use detected default branch in all worktree tests --- tests/core/test_worktrees.py | 15 +++++++++------ 1 file changed, 9 insertions(+), 6 deletions(-) diff --git a/tests/core/test_worktrees.py b/tests/core/test_worktrees.py index 55aefb39..3a5890b0 100644 --- a/tests/core/test_worktrees.py +++ b/tests/core/test_worktrees.py @@ -52,25 +52,26 @@ class TestTaskWorktreeCreate: def test_creates_worktree(self, tmp_path: Path) -> None: from codeframe.core.worktrees import TaskWorktree - # Set up a real git repo subprocess.run(["git", "init", str(tmp_path)], capture_output=True) subprocess.run(["git", "-C", str(tmp_path), "commit", "--allow-empty", "-m", "init"], capture_output=True) + base = _get_default_branch(tmp_path) wt = TaskWorktree() - worktree_path = wt.create(tmp_path, "task-1") + worktree_path = wt.create(tmp_path, "task-1", base_branch=base) assert worktree_path.exists() assert worktree_path.name == "task-1" - assert (worktree_path / ".git").exists() # worktrees have a .git file + assert (worktree_path / ".git").exists() def test_returns_correct_path(self, tmp_path: Path) -> None: from codeframe.core.worktrees import TaskWorktree subprocess.run(["git", "init", str(tmp_path)], capture_output=True) subprocess.run(["git", "-C", str(tmp_path), "commit", "--allow-empty", "-m", "init"], capture_output=True) + base = _get_default_branch(tmp_path) wt = TaskWorktree() - path = wt.create(tmp_path, "my-task") + path = wt.create(tmp_path, "my-task", base_branch=base) expected = tmp_path / ".codeframe" / "worktrees" / "my-task" assert path == expected @@ -80,9 +81,10 @@ def test_creates_branch_with_cf_prefix(self, tmp_path: Path) -> None: subprocess.run(["git", "init", str(tmp_path)], capture_output=True) subprocess.run(["git", "-C", str(tmp_path), "commit", "--allow-empty", "-m", "init"], capture_output=True) + base = _get_default_branch(tmp_path) wt = TaskWorktree() - wt.create(tmp_path, "task-1") + wt.create(tmp_path, "task-1", base_branch=base) # Check branch exists result = subprocess.run( @@ -155,9 +157,10 @@ def test_removes_worktree_and_branch(self, tmp_path: Path) -> None: subprocess.run(["git", "init", str(tmp_path)], capture_output=True) subprocess.run(["git", "-C", str(tmp_path), "commit", "--allow-empty", "-m", "init"], capture_output=True) + base = _get_default_branch(tmp_path) wt = TaskWorktree() - worktree_path = wt.create(tmp_path, "task-1") + worktree_path = wt.create(tmp_path, "task-1", base_branch=base) assert worktree_path.exists() wt.cleanup(tmp_path, "task-1")