fix: serialize git worktree mutations (#111)

* fix: serialize git worktree mutations

* fix: address worktree review nits

Author: xiaojiou176

apps/orchestrator/src/openvibecoding_orch/worktrees/manager.py

@@ -1,11 +1,22 @@
 from __future__ import annotations
 
+from contextlib import contextmanager
 import re
 import subprocess
+import threading
 from pathlib import Path
 
 from openvibecoding_orch.config import load_config
 
+try:
+    import fcntl
+except Exception:  # pragma: no cover - non-posix fallback
+    fcntl = None
+
+
+_PROCESS_LOCK_GUARD = threading.Lock()
+_PROCESS_LOCKS: dict[str, threading.RLock] = {}
+
 
 def _run_git(args: list[str], repo_root: Path) -> subprocess.CompletedProcess[str]:
     return subprocess.run(
@@ -27,6 +38,33 @@ def _repo_root() -> Path:
     return Path(result.stdout.strip())
 
 
+def _process_lock_for(path: Path) -> threading.RLock:
+    key = str(path.resolve(strict=False))
+    with _PROCESS_LOCK_GUARD:
+        lock = _PROCESS_LOCKS.get(key)
+        if lock is None:
+            lock = threading.RLock()
+            _PROCESS_LOCKS[key] = lock
+        return lock
+
+
+@contextmanager
+def _worktree_file_lock(repo_root: Path):
+    lock_path = repo_root / ".git" / "openvibecoding-worktrees.lock"
+    lock_path.parent.mkdir(parents=True, exist_ok=True)
+    process_lock = _process_lock_for(lock_path)
+    with process_lock:
+        if fcntl is None:
+            yield None
+            return
+        with lock_path.open("a+", encoding="utf-8") as lock_file:
+            fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX)
+            try:
+                yield lock_file
+            finally:
+                fcntl.flock(lock_file.fileno(), fcntl.LOCK_UN)
+
+
 def _safe_segment(value: str) -> str:
     cleaned = re.sub(r"[^A-Za-z0-9._-]+", "-", value.strip())
     return cleaned or "task"
@@ -45,66 +83,69 @@ def _is_stale_worktree_metadata_error(stderr: str) -> bool:
 def create_worktree(run_id: str, task_id: str, baseline_commit: str) -> Path:
     cfg = load_config()
     repo_root = _repo_root()
+    with _worktree_file_lock(repo_root):
+        prune = _run_git(["git", "worktree", "prune"], repo_root)
+        _ensure_git_ok(prune, ["git", "worktree", "prune"])
 
-    prune = _run_git(["git", "worktree", "prune"], repo_root)
-    _ensure_git_ok(prune, ["git", "worktree", "prune"])
+        check = _run_git(["git", "cat-file", "-e", baseline_commit], repo_root)
+        _ensure_git_ok(check, ["git", "cat-file", "-e", baseline_commit])
 
-    check = _run_git(["git", "cat-file", "-e", baseline_commit], repo_root)
-    _ensure_git_ok(check, ["git", "cat-file", "-e", baseline_commit])
+        safe_task = _safe_segment(task_id)
+        worktree_path = cfg.worktree_root / run_id / safe_task
+        worktree_path.parent.mkdir(parents=True, exist_ok=True)
+        branch = _branch_name(run_id, task_id)
 
-    safe_task = _safe_segment(task_id)
-    worktree_path = cfg.worktree_root / run_id / safe_task
-    worktree_path.parent.mkdir(parents=True, exist_ok=True)
-    branch = _branch_name(run_id, task_id)
-
-    cmd = [
-        "git",
-        "worktree",
-        "add",
-        "-b",
-        branch,
-        str(worktree_path),
-        baseline_commit,
-    ]
-    result = _run_git(cmd, repo_root)
-    _ensure_git_ok(result, cmd)
-    return worktree_path
+        cmd = [
+            "git",
+            "worktree",
+            "add",
+            "-b",
+            branch,
+            str(worktree_path),
+            baseline_commit,
+        ]
+        result = _run_git(cmd, repo_root)
+        _ensure_git_ok(result, cmd)
+        return worktree_path
 
 
 def remove_worktree(run_id: str, task_id: str) -> None:
     cfg = load_config()
     repo_root = _repo_root()
-    safe_task = _safe_segment(task_id)
-    worktree_path = cfg.worktree_root / run_id / safe_task
-    branch = _branch_name(run_id, task_id)
-
-    if worktree_path.exists():
-        cmd_remove = ["git", "worktree", "remove", "--force", str(worktree_path)]
-        result = _run_git(cmd_remove, repo_root)
-        _ensure_git_ok(result, cmd_remove)
-
-    prune = _run_git(["git", "worktree", "prune"], repo_root)
-    _ensure_git_ok(prune, ["git", "worktree", "prune"])
-
-    parent = worktree_path.parent
-    if parent.exists() and parent.is_dir():
-        try:
-            parent.rmdir()
-        except OSError:
-            pass
-
-    cmd_branch = ["git", "branch", "-D", branch]
-    result = _run_git(cmd_branch, repo_root)
-    if result.returncode != 0 and _is_stale_worktree_metadata_error(result.stderr):
+    with _worktree_file_lock(repo_root):
+        safe_task = _safe_segment(task_id)
+        worktree_path = cfg.worktree_root / run_id / safe_task
+        branch = _branch_name(run_id, task_id)
+
+        if worktree_path.exists():
+            cmd_remove = ["git", "worktree", "remove", "--force", str(worktree_path)]
+            result = _run_git(cmd_remove, repo_root)
+            _ensure_git_ok(result, cmd_remove)
+
         prune = _run_git(["git", "worktree", "prune"], repo_root)
         _ensure_git_ok(prune, ["git", "worktree", "prune"])
+
+        parent = worktree_path.parent
+        if parent.exists() and parent.is_dir():
+            try:
+                parent.rmdir()
+            except OSError:
+                # Best-effort cleanup only; parent may still hold other task worktrees.
+                pass
+
+        cmd_branch = ["git", "branch", "-D", branch]
         result = _run_git(cmd_branch, repo_root)
-    if result.returncode != 0 and "not found" not in result.stderr:
-        _ensure_git_ok(result, cmd_branch)
+        if result.returncode != 0 and _is_stale_worktree_metadata_error(result.stderr):
+            prune = _run_git(["git", "worktree", "prune"], repo_root)
+            _ensure_git_ok(prune, ["git", "worktree", "prune"])
+            result = _run_git(cmd_branch, repo_root)
+        if result.returncode != 0 and "not found" not in result.stderr:
+            _ensure_git_ok(result, cmd_branch)
 
 
 def list_worktrees() -> list[str]:
     repo_root = _repo_root()
-    result = _run_git(["git", "worktree", "list", "--porcelain"], repo_root)
-    _ensure_git_ok(result, ["git", "worktree", "list", "--porcelain"])
-    return [line for line in result.stdout.splitlines() if line.startswith("worktree ")]
+    with _worktree_file_lock(repo_root):
+        result = _run_git(["git", "worktree", "list", "--porcelain"], repo_root)
+        _ensure_git_ok(result, ["git", "worktree", "list", "--porcelain"])
+        return [line for line in result.stdout.splitlines() if line.startswith("worktree ")]

apps/orchestrator/tests/test_worktree_lock.py

@@ -1,5 +1,7 @@
 import os
 import subprocess
+import threading
+import time
 from pathlib import Path
 
 from openvibecoding_orch.locks import locker
@@ -70,3 +72,57 @@ def test_locks_atomic(tmp_path: Path):
     os.environ["OPENVIBECODING_RUN_ID"] = "run_b"
     assert locker.acquire_lock(target) is True
     locker.release_lock(target)
+
+
+def test_worktree_create_serializes_parallel_mutations(tmp_path: Path, monkeypatch) -> None:
+    repo = tmp_path / "repo_lock"
+    repo.mkdir(parents=True)
+    _init_repo(repo)
+    (repo / "README.md").write_text("hello", encoding="utf-8")
+    _git(["git", "add", "README.md"], cwd=repo)
+    _git(["git", "commit", "-m", "init"], cwd=repo)
+    baseline = _git_output(["git", "rev-parse", "HEAD"], cwd=repo).strip()
+
+    os.environ["OPENVIBECODING_WORKTREE_ROOT"] = str(tmp_path / "worktrees")
+    cwd = Path.cwd()
+    active_mutation_calls = 0
+    overlap_detected = False
+    call_guard = threading.Lock()
+
+    def _fake_run_git(args: list[str], repo_root: Path):
+        nonlocal active_mutation_calls, overlap_detected
+        if args[:3] == ["git", "rev-parse", "--show-toplevel"]:
+            return subprocess.CompletedProcess(args, 0, str(repo) + "\n", "")
+        if args[:3] == ["git", "cat-file", "-e"]:
+            return subprocess.CompletedProcess(args, 0, "", "")
+        if args[:3] == ["git", "worktree", "prune"] or args[:3] == ["git", "worktree", "add"]:
+            with call_guard:
+                active_mutation_calls += 1
+                if active_mutation_calls > 1:
+                    overlap_detected = True
+            try:
+                time.sleep(0.05)
+                if args[:3] == ["git", "worktree", "add"]:
+                    Path(args[5]).mkdir(parents=True, exist_ok=True)
+                return subprocess.CompletedProcess(args, 0, "", "")
+            finally:
+                with call_guard:
+                    active_mutation_calls -= 1
+        return subprocess.CompletedProcess(args, 0, "", "")
+
+    monkeypatch.setattr(manager, "_run_git", _fake_run_git)
+
+    try:
+        os.chdir(repo)
+        threads = [
+            threading.Thread(target=manager.create_worktree, args=("run-a", "worker_core_01", baseline)),
+            threading.Thread(target=manager.create_worktree, args=("run-b", "worker_frontend_01", baseline)),
+        ]
+        for thread in threads:
+            thread.start()
+        for thread in threads:
+            thread.join()
+    finally:
+        os.chdir(cwd)
+
+    assert overlap_detected is False