Remediate PR #853 in response to review #525 for Issue #850

author_issue_bootstrap.py

@@ -302,6 +302,39 @@ def assess_author_issue_bootstrap(
     }
 
 
+import fcntl
+
+
+class BootstrapTransitionLock:
+    """Inter-process file lock scoped to the transition identity / idempotency key."""
+
+    def __init__(self, idempotency_key: str, journal_dir: str | None = None):
+        safe_key = "".join(
+            c if c.isalnum() or c in ("-", "_", ".") else "_"
+            for c in idempotency_key
+        )
+        lock_dir = get_journal_dir(journal_dir)
+        self.lock_path = os.path.join(lock_dir, f"{safe_key}.lock")
+        self.fd = None
+
+    def __enter__(self):
+        self.fd = open(self.lock_path, "a+")
+        fcntl.flock(self.fd, fcntl.LOCK_EX)
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        if self.fd:
+            try:
+                fcntl.flock(self.fd, fcntl.LOCK_UN)
+            except Exception:
+                pass
+            try:
+                self.fd.close()
+            except Exception:
+                pass
+            self.fd = None
+
+
 def bootstrap_author_issue_worktree(
     *,
     issue_number: int,
@@ -359,418 +392,433 @@ def bootstrap_author_issue_worktree(
             lease_id=lease_id,
         )
 
-    # Idempotency check
+    # Acquire cross-process file lock scoped to the idempotency key / transition identity
-    existing = load_phase_journal(key)
+    with BootstrapTransitionLock(key):
-    if existing and existing.get("completed"):
+        # Idempotency check
-        if (
+        existing = load_phase_journal(key)
-            existing.get("issue_number") == issue_number
+        if existing and existing.get("completed"):
-            and existing.get("branch_name") == target_branch
+            if (
-            and os.path.realpath(existing.get("worktree_path", ""))
+                existing.get("issue_number") == issue_number
-            == target_worktree
+                and existing.get("branch_name") == target_branch
-        ):
+                and os.path.realpath(existing.get("worktree_path", ""))
+                == target_worktree
+            ):
+                return {
+                    "success": True,
+                    "replayed": True,
+                    "message": (
+                        f"Idempotent replay: worktree for issue #{issue_number} already bootstrapped at {target_worktree}"
+                    ),
+                    "issue_number": issue_number,
+                    "branch_name": target_branch,
+                    "worktree_path": target_worktree,
+                    "base_sha": existing.get("resolved_base_sha"),
+                    "lease_id": existing.get("lease_id"),
+                    "assignment_id": existing.get("assignment_id"),
+                    "idempotency_key": key,
+                    "phase_journal": existing,
+                    "exact_next_action": (
+                        f"Call gitea_whoami, then gitea_resolve_task_capability(task='work_issue', worktree_path='{target_worktree}') "
+                        "and proceed with author implementation in the bootstrapped worktree."
+                    ),
+                }
+            else:
+                return {
+                    "success": False,
+                    "reason_code": "incompatible_idempotency_replay",
+                    "message": (
+                        f"Idempotency key '{key}' already exists with incompatible parameters "
+                        f"(stored: {existing.get('branch_name')}, {existing.get('worktree_path')}; "
+                        f"requested: {target_branch}, {target_worktree})"
+                    ),
+                    "exact_next_action": (
+                        "Supply a unique idempotency_key or pass compatible parameters."
+                    ),
+                }
+
+        # Initialize or resume Phase Journal
+        if existing:
+            journal = existing
+            artifacts = journal.setdefault("artifacts_created", {})
+            artifacts.setdefault("branch_created", False)
+            artifacts.setdefault("worktree_dir_created", False)
+            artifacts.setdefault("worktree_registered", False)
+            artifacts.setdefault("lock_created", False)
+        else:
+            journal = {
+                "idempotency_key": key,
+                "issue_number": issue_number,
+                "assignment_id": assignment_id,
+                "lease_id": lease_id,
+                "expected_base_sha": expected_base_sha,
+                "resolved_base_sha": None,
+                "branch_name": target_branch,
+                "worktree_path": target_worktree,
+                "active_identity": active_identity,
+                "active_profile": active_profile,
+                "owner_session": owner_session,
+                "remote": remote,
+                "org": org,
+                "repo": repo,
+                "phases": {},
+                "artifacts_created": {
+                    "branch_created": False,
+                    "worktree_dir_created": False,
+                    "worktree_registered": False,
+                    "lock_created": False,
+                },
+                "current_phase": PHASE_1_REQUEST_ACCEPTED,
+                "completed": False,
+            }
+
+        # Fetch current live master SHA
+        try:
+            rev_res = subprocess.run(
+                ["git", "-C", root, "rev-parse", "HEAD"],
+                capture_output=True,
+                text=True,
+                check=True,
+            )
+            live_master_sha = rev_res.stdout.strip()
+        except Exception as exc:
+            return {
+                "success": False,
+                "reason_code": "git_rev_parse_failed",
+                "message": f"Could not determine repository HEAD: {exc}",
+                "exact_next_action": "Verify repository git state and retry.",
+            }
+
+        # Phase 1: REQUEST_ACCEPTED & Concurrency Pin Check
+        if expected_base_sha:
+            exp_norm = expected_base_sha.strip().lower()
+            live_norm = live_master_sha.lower()
+            if exp_norm != live_norm:
+                journal["failure_reason"] = (
+                    f"stale concurrency pin: expected {exp_norm[:12]} != live {live_norm[:12]}"
+                )
+                save_phase_journal(journal)
+                return {
+                    "success": False,
+                    "reason_code": "stale_concurrency_pin",
+                    "message": (
+                        f"Expected base SHA {exp_norm[:12]} does not match live master SHA {live_norm[:12]} (fail closed)."
+                    ),
+                    "expected_base_sha": expected_base_sha,
+                    "live_master_sha": live_master_sha,
+                    "exact_next_action": (
+                        "Re-evaluate assignment against current live master SHA and retry with updated expected_base_sha."
+                    ),
+                }
+
+        journal["resolved_base_sha"] = live_master_sha
+        journal["phases"][PHASE_1_REQUEST_ACCEPTED] = {
+            "status": "completed",
+            "live_master_sha": live_master_sha,
+            "expected_base_sha": expected_base_sha,
+        }
+        journal["current_phase"] = PHASE_2_BRANCH_CONFIRMED
+        save_phase_journal(journal)
+
+        if dry_run:
             return {
                 "success": True,
-                "replayed": True,
+                "dry_run": True,
-                "message": (
+                "message": f"Dry-run: validated bootstrap intent for issue #{issue_number}",
-                    f"Idempotent replay: worktree for issue #{issue_number} already bootstrapped at {target_worktree}"
-                ),
                 "issue_number": issue_number,
                 "branch_name": target_branch,
                 "worktree_path": target_worktree,
-                "base_sha": existing.get("resolved_base_sha"),
+                "base_sha": live_master_sha,
-                "lease_id": existing.get("lease_id"),
+                "phase_journal": journal,
-                "assignment_id": existing.get("assignment_id"),
+                "exact_next_action": "Run without dry_run=True to execute bootstrap.",
-                "idempotency_key": key,
-                "phase_journal": existing,
-                "exact_next_action": (
-                    f"Call gitea_whoami, then gitea_resolve_task_capability(task='work_issue', worktree_path='{target_worktree}') "
-                    "and proceed with author implementation in the bootstrapped worktree."
-                ),
-            }
-        else:
-            return {
-                "success": False,
-                "reason_code": "incompatible_idempotency_replay",
-                "message": (
-                    f"Idempotency key '{key}' already exists with incompatible parameters "
-                    f"(stored: {existing.get('branch_name')}, {existing.get('worktree_path')}; "
-                    f"requested: {target_branch}, {target_worktree})"
-                ),
-                "exact_next_action": (
-                    "Supply a unique idempotency_key or pass compatible parameters."
-                ),
             }
 
-    # Initialize Phase Journal
+        # Phase 2: BRANCH_CONFIRMED
-    journal = existing or {
+        was_branch_created_previously = journal["artifacts_created"].get("branch_created", False)
-        "idempotency_key": key,
+        branch_check = subprocess.run(
-        "issue_number": issue_number,
+            ["git", "-C", root, "rev-parse", "--verify", target_branch],
-        "assignment_id": assignment_id,
-        "lease_id": lease_id,
-        "expected_base_sha": expected_base_sha,
-        "resolved_base_sha": None,
-        "branch_name": target_branch,
-        "worktree_path": target_worktree,
-        "active_identity": active_identity,
-        "active_profile": active_profile,
-        "owner_session": owner_session,
-        "remote": remote,
-        "org": org,
-        "repo": repo,
-        "phases": {},
-        "artifacts_created": {
-            "branch_created": False,
-            "worktree_dir_created": False,
-            "worktree_registered": False,
-            "lock_created": False,
-        },
-        "current_phase": PHASE_1_REQUEST_ACCEPTED,
-        "completed": False,
-    }
-
-    # Fetch current live master SHA
-    try:
-        rev_res = subprocess.run(
-            ["git", "-C", root, "rev-parse", "HEAD"],
             capture_output=True,
             text=True,
-            check=True,
+            check=False,
         )
-        live_master_sha = rev_res.stdout.strip()
+        if branch_check.returncode == 0:
-    except Exception as exc:
+            branch_head = branch_check.stdout.strip()
-        return {
+            # Verify branch head descends from base
-            "success": False,
+            anc_check = subprocess.run(
-            "reason_code": "git_rev_parse_failed",
+                [
-            "message": f"Could not determine repository HEAD: {exc}",
+                    "git",
-            "exact_next_action": "Verify repository git state and retry.",
+                    "-C",
-        }
+                    root,
-
+                    "merge-base",
-    # Phase 1: REQUEST_ACCEPTED & Concurrency Pin Check
+                    "--is-ancestor",
-    if expected_base_sha:
+                    live_master_sha,
-        exp_norm = expected_base_sha.strip().lower()
+                    branch_head,
-        live_norm = live_master_sha.lower()
+                ],
-        if exp_norm != live_norm:
+                capture_output=True,
-            journal["failure_reason"] = (
+                text=True,
-                f"stale concurrency pin: expected {exp_norm[:12]} != live {live_norm[:12]}"
+                check=False,
             )
-            save_phase_journal(journal)
+            if anc_check.returncode != 0 and branch_head.lower() != live_master_sha.lower():
+                journal["failure_reason"] = (
+                    f"existing branch '{target_branch}' HEAD ({branch_head[:12]}) does not descend from base ({live_master_sha[:12]})"
+                )
+                save_phase_journal(journal)
+                return {
+                    "success": False,
+                    "reason_code": "incompatible_existing_branch",
+                    "message": (
+                        f"Existing branch '{target_branch}' HEAD ({branch_head[:12]}) is incompatible with live master ({live_master_sha[:12]})."
+                    ),
+                    "exact_next_action": (
+                        "Inspect or remove the incompatible branch before bootstrapping."
+                    ),
+                }
+            # Preserve creation provenance monotonically across interruption and replay
+            journal["artifacts_created"]["branch_created"] = was_branch_created_previously
+        else:
+            # Create branch
+            create_res = subprocess.run(
+                ["git", "-C", root, "branch", target_branch, live_master_sha],
+                capture_output=True,
+                text=True,
+                check=False,
+            )
+            if create_res.returncode != 0:
+                journal["failure_reason"] = (
+                    f"failed to create git branch '{target_branch}': {create_res.stderr.strip()}"
+                )
+                save_phase_journal(journal)
+                return {
+                    "success": False,
+                    "reason_code": "branch_creation_failed",
+                    "message": f"Failed to create git branch '{target_branch}': {create_res.stderr.strip()}",
+                    "exact_next_action": "Verify branch availability and retry.",
+                }
+            journal["artifacts_created"]["branch_created"] = True
+
+        journal["phases"][PHASE_2_BRANCH_CONFIRMED] = {
+            "status": "completed",
+            "branch_name": target_branch,
+            "created": journal["artifacts_created"]["branch_created"],
+        }
+        journal["current_phase"] = PHASE_3_PATH_RESERVED
+        save_phase_journal(journal)
+
+        # Phase 3: PATH_RESERVED & Phase 4: WORKTREE_CONFIRMED
+        if not author_mutation_worktree.is_path_under_branches(
+            target_worktree, root
+        ):
+            journal["failure_reason"] = (
+                f"target_worktree '{target_worktree}' is outside canonical branches/ root"
+            )
+            run_compensating_recovery(journal, root)
             return {
                 "success": False,
-                "reason_code": "stale_concurrency_pin",
+                "reason_code": "path_outside_canonical_branches_root",
                 "message": (
-                    f"Expected base SHA {exp_norm[:12]} does not match live master SHA {live_norm[:12]} (fail closed)."
+                    f"Worktree path '{target_worktree}' is outside canonical branches/ root (fail closed)."
                 ),
-                "expected_base_sha": expected_base_sha,
-                "live_master_sha": live_master_sha,
                 "exact_next_action": (
-                    "Re-evaluate assignment against current live master SHA and retry with updated expected_base_sha."
+                    "Provide a worktree_path inside canonical branches/ root, e.g., 'branches/issue-...'"
                 ),
             }
 
-    journal["resolved_base_sha"] = live_master_sha
+        was_dir_created_previously = journal["artifacts_created"].get("worktree_dir_created", False)
-    journal["phases"][PHASE_1_REQUEST_ACCEPTED] = {
+        was_registered_previously = journal["artifacts_created"].get("worktree_registered", False)
-        "status": "completed",
+
-        "live_master_sha": live_master_sha,
+        dir_exists = os.path.exists(target_worktree)
-        "expected_base_sha": expected_base_sha,
+        if dir_exists:
-    }
+            # Check porcelain directly
-    journal["current_phase"] = PHASE_2_BRANCH_CONFIRMED
+            porc_res = subprocess.run(
-    save_phase_journal(journal)
+                ["git", "-C", target_worktree, "status", "--porcelain"],
+                capture_output=True,
+                text=True,
+                check=False,
+            )
+            dirty = (
+                parse_dirty_tracked_files(porc_res.stdout)
+                if porc_res.returncode == 0
+                else []
+            )
+            if dirty or (porc_res.returncode == 0 and porc_res.stdout.strip()):
+                journal["failure_reason"] = (
+                    f"target worktree '{target_worktree}' contains dirty tracked/untracked files"
+                )
+                run_compensating_recovery(journal, root)
+                return {
+                    "success": False,
+                    "reason_code": "preexisting_dirty_worktree",
+                    "message": (
+                        f"Preexisting worktree '{target_worktree}' has dirty tracked/untracked files (fail closed)."
+                    ),
+                    "exact_next_action": (
+                        "Clean or stash the pre-existing worktree files before bootstrapping."
+                    ),
+                }
+
+            # Check registered branch
+            wt_state = issue_lock_worktree.read_worktree_git_state(target_worktree)
+            wt_branch = (wt_state.get("current_branch") or "").strip()
+            if wt_branch and wt_branch != target_branch:
+                journal["failure_reason"] = (
+                    f"existing worktree '{target_worktree}' is on branch '{wt_branch}' != expected '{target_branch}'"
+                )
+                run_compensating_recovery(journal, root)
+                return {
+                    "success": False,
+                    "reason_code": "incompatible_existing_directory",
+                    "message": (
+                        f"Existing worktree '{target_worktree}' is registered to branch '{wt_branch}' instead of '{target_branch}'."
+                    ),
+                    "exact_next_action": (
+                        "Inspect or remove the pre-existing worktree folder before bootstrapping."
+                    ),
+                }
+            journal["artifacts_created"]["worktree_dir_created"] = was_dir_created_previously
+            journal["artifacts_created"]["worktree_registered"] = (
+                was_registered_previously or was_dir_created_previously
+            )
+        else:
+            wt_add_res = subprocess.run(
+                [
+                    "git",
+                    "-C",
+                    root,
+                    "worktree",
+                    "add",
+                    target_worktree,
+                    target_branch,
+                ],
+                capture_output=True,
+                text=True,
+                check=False,
+            )
+            if wt_add_res.returncode != 0:
+                journal["failure_reason"] = (
+                    f"git worktree add failed: {wt_add_res.stderr.strip()}"
+                )
+                run_compensating_recovery(journal, root)
+                return {
+                    "success": False,
+                    "reason_code": "worktree_add_failed",
+                    "message": f"Failed to execute git worktree add: {wt_add_res.stderr.strip()}",
+                    "exact_next_action": "Verify git worktree capabilities and retry.",
+                }
+            journal["artifacts_created"]["worktree_dir_created"] = True
+            journal["artifacts_created"]["worktree_registered"] = True
+
+        journal["phases"][PHASE_3_PATH_RESERVED] = {
+            "status": "completed",
+            "worktree_path": target_worktree,
+            "preexisting_dir": dir_exists,
+        }
+        journal["current_phase"] = PHASE_4_WORKTREE_CONFIRMED
+        save_phase_journal(journal)
+
+        # Phase 5: REGISTRATION_VERIFIED
+        wt_list_res = subprocess.run(
+            ["git", "-C", root, "worktree", "list", "--porcelain"],
+            capture_output=True,
+            text=True,
+            check=False,
+        )
+        norm_target = os.path.realpath(target_worktree)
+        found_registration = False
+        if wt_list_res.returncode == 0:
+            for block in wt_list_res.stdout.split("\n\n"):
+                lines = block.strip().splitlines()
+                worktree_line = next(
+                    (l[9:].strip() for l in lines if l.startswith("worktree ")),
+                    None,
+                )
+                if worktree_line and os.path.realpath(worktree_line) == norm_target:
+                    found_registration = True
+                    break
+
+        if not found_registration:
+            journal["failure_reason"] = (
+                f"worktree registration for '{target_worktree}' not found in git worktree list"
+            )
+            run_compensating_recovery(journal, root)
+            return {
+                "success": False,
+                "reason_code": "worktree_registration_verification_failed",
+                "message": f"Worktree '{target_worktree}' registration verification failed.",
+                "exact_next_action": "Check git worktree list integrity and retry.",
+            }
+
+        journal["phases"][PHASE_4_WORKTREE_CONFIRMED] = {
+            "status": "completed",
+            "worktree_path": target_worktree,
+        }
+        journal["phases"][PHASE_5_REGISTRATION_VERIFIED] = {
+            "status": "completed",
+            "registered": True,
+        }
+        journal["current_phase"] = PHASE_6_STATE_ESTABLISHED
+        save_phase_journal(journal)
+
+        # Phase 6: STATE_ESTABLISHED — Issue Lock Acquisition
+        from datetime import datetime, timezone
+        try:
+            lock_data = {
+                "remote": remote,
+                "org": org or "Scaled-Tech-Consulting",
+                "repo": repo or "Gitea-Tools",
+                "issue_number": issue_number,
+                "branch": target_branch,
+                "branch_name": target_branch,
+                "worktree_path": target_worktree,
+                "owner_session": owner_session or "prgs-author-95048-63667752",
+                "claimant": {
+                    "username": active_identity or "jcwalker3",
+                    "profile": active_profile or "prgs-author",
+                },
+                "assignment_id": assignment_id,
+                "lease_id": lease_id,
+                "expected_base_sha": live_master_sha,
+                "created_at": datetime.now(timezone.utc).isoformat(),
+            }
+            lock_res = issue_lock_store.bind_session_lock(lock_data, lock_dir=lock_dir)
+            journal["artifacts_created"]["lock_created"] = True
+        except Exception as exc:
+            journal["failure_reason"] = f"issue lock binding failed: {exc}"
+            run_compensating_recovery(journal, root)
+            return {
+                "success": False,
+                "reason_code": "issue_lock_acquisition_failed",
+                "message": f"Could not bind canonical issue lock for issue #{issue_number}: {exc}",
+                "exact_next_action": "Verify lease/assignment state and retry.",
+            }
+
+        journal["phases"][PHASE_6_STATE_ESTABLISHED] = {
+            "status": "completed",
+            "lock": lock_res,
+        }
+        journal["phases"][PHASE_7_TRANSITION_COMPLETED] = {
+            "status": "completed",
+        }
+        journal["current_phase"] = PHASE_7_TRANSITION_COMPLETED
+        journal["completed"] = True
+        save_phase_journal(journal)
 
-    if dry_run:
         return {
             "success": True,
-            "dry_run": True,
+            "replayed": False,
-            "message": f"Dry-run: validated bootstrap intent for issue #{issue_number}",
+            "message": (
+                f"Successfully bootstrapped author issue worktree for issue #{issue_number} "
+                f"at branch '{target_branch}' and worktree '{target_worktree}'."
+            ),
             "issue_number": issue_number,
             "branch_name": target_branch,
             "worktree_path": target_worktree,
             "base_sha": live_master_sha,
-            "phase_journal": journal,
-            "exact_next_action": "Run without dry_run=True to execute bootstrap.",
-        }
-
-    # Phase 2: BRANCH_CONFIRMED
-    branch_check = subprocess.run(
-        ["git", "-C", root, "rev-parse", "--verify", target_branch],
-        capture_output=True,
-        text=True,
-        check=False,
-    )
-    if branch_check.returncode == 0:
-        branch_head = branch_check.stdout.strip()
-        # Verify branch head descends from base
-        anc_check = subprocess.run(
-            [
-                "git",
-                "-C",
-                root,
-                "merge-base",
-                "--is-ancestor",
-                live_master_sha,
-                branch_head,
-            ],
-            capture_output=True,
-            text=True,
-            check=False,
-        )
-        if anc_check.returncode != 0 and branch_head.lower() != live_master_sha.lower():
-            journal["failure_reason"] = (
-                f"existing branch '{target_branch}' HEAD ({branch_head[:12]}) does not descend from base ({live_master_sha[:12]})"
-            )
-            save_phase_journal(journal)
-            return {
-                "success": False,
-                "reason_code": "incompatible_existing_branch",
-                "message": (
-                    f"Existing branch '{target_branch}' HEAD ({branch_head[:12]}) is incompatible with live master ({live_master_sha[:12]})."
-                ),
-                "exact_next_action": (
-                    "Inspect or remove the incompatible branch before bootstrapping."
-                ),
-            }
-        journal["artifacts_created"]["branch_created"] = False
-    else:
-        # Create branch
-        create_res = subprocess.run(
-            ["git", "-C", root, "branch", target_branch, live_master_sha],
-            capture_output=True,
-            text=True,
-            check=False,
-        )
-        if create_res.returncode != 0:
-            journal["failure_reason"] = (
-                f"failed to create git branch '{target_branch}': {create_res.stderr.strip()}"
-            )
-            save_phase_journal(journal)
-            return {
-                "success": False,
-                "reason_code": "branch_creation_failed",
-                "message": f"Failed to create git branch '{target_branch}': {create_res.stderr.strip()}",
-                "exact_next_action": "Verify branch availability and retry.",
-            }
-        journal["artifacts_created"]["branch_created"] = True
-
-    journal["phases"][PHASE_2_BRANCH_CONFIRMED] = {
-        "status": "completed",
-        "branch_name": target_branch,
-        "created": journal["artifacts_created"]["branch_created"],
-    }
-    journal["current_phase"] = PHASE_3_PATH_RESERVED
-    save_phase_journal(journal)
-
-    # Phase 3: PATH_RESERVED
-    if not author_mutation_worktree.is_path_under_branches(
-        target_worktree, root
-    ):
-        journal["failure_reason"] = (
-            f"target_worktree '{target_worktree}' is outside canonical branches/ root"
-        )
-        run_compensating_recovery(journal, root)
-        return {
-            "success": False,
-            "reason_code": "path_outside_canonical_branches_root",
-            "message": (
-                f"Worktree path '{target_worktree}' is outside canonical branches/ root (fail closed)."
-            ),
-            "exact_next_action": (
-                "Provide a worktree_path inside canonical branches/ root, e.g., 'branches/issue-...'"
-            ),
-        }
-
-    dir_exists = os.path.exists(target_worktree)
-    if dir_exists:
-        # Check porcelain directly
-        porc_res = subprocess.run(
-            ["git", "-C", target_worktree, "status", "--porcelain"],
-            capture_output=True,
-            text=True,
-            check=False,
-        )
-        dirty = (
-            parse_dirty_tracked_files(porc_res.stdout)
-            if porc_res.returncode == 0
-            else []
-        )
-        if dirty or (porc_res.returncode == 0 and porc_res.stdout.strip()):
-            journal["failure_reason"] = (
-                f"target worktree '{target_worktree}' contains dirty tracked/untracked files"
-            )
-            run_compensating_recovery(journal, root)
-            return {
-                "success": False,
-                "reason_code": "preexisting_dirty_worktree",
-                "message": (
-                    f"Preexisting worktree '{target_worktree}' has dirty tracked/untracked files (fail closed)."
-                ),
-                "exact_next_action": (
-                    "Clean or stash the pre-existing worktree files before bootstrapping."
-                ),
-            }
-
-        # Check registered branch
-        wt_state = issue_lock_worktree.read_worktree_git_state(target_worktree)
-        wt_branch = (wt_state.get("current_branch") or "").strip()
-        if wt_branch and wt_branch != target_branch:
-            journal["failure_reason"] = (
-                f"existing worktree '{target_worktree}' is on branch '{wt_branch}' != expected '{target_branch}'"
-            )
-            run_compensating_recovery(journal, root)
-            return {
-                "success": False,
-                "reason_code": "incompatible_existing_directory",
-                "message": (
-                    f"Existing worktree '{target_worktree}' is registered to branch '{wt_branch}' instead of '{target_branch}'."
-                ),
-                "exact_next_action": (
-                    "Inspect or remove the pre-existing worktree folder before bootstrapping."
-                ),
-            }
-        journal["artifacts_created"]["worktree_dir_created"] = False
-    else:
-        journal["artifacts_created"]["worktree_dir_created"] = True
-
-    journal["phases"][PHASE_3_PATH_RESERVED] = {
-        "status": "completed",
-        "worktree_path": target_worktree,
-        "preexisting_dir": dir_exists,
-    }
-    journal["current_phase"] = PHASE_4_WORKTREE_CONFIRMED
-    save_phase_journal(journal)
-
-    # Phase 4: WORKTREE_CONFIRMED & Phase 5: REGISTRATION_VERIFIED
-    if journal["artifacts_created"]["worktree_dir_created"]:
-        wt_add_res = subprocess.run(
-            [
-                "git",
-                "-C",
-                root,
-                "worktree",
-                "add",
-                target_worktree,
-                target_branch,
-            ],
-            capture_output=True,
-            text=True,
-            check=False,
-        )
-        if wt_add_res.returncode != 0:
-            journal["failure_reason"] = (
-                f"git worktree add failed: {wt_add_res.stderr.strip()}"
-            )
-            run_compensating_recovery(journal, root)
-            return {
-                "success": False,
-                "reason_code": "worktree_add_failed",
-                "message": f"Failed to execute git worktree add: {wt_add_res.stderr.strip()}",
-                "exact_next_action": "Verify git worktree capabilities and retry.",
-            }
-        journal["artifacts_created"]["worktree_registered"] = True
-
-    # Verify registration in git worktree list
-    wt_list_res = subprocess.run(
-        ["git", "-C", root, "worktree", "list", "--porcelain"],
-        capture_output=True,
-        text=True,
-        check=False,
-    )
-    norm_target = os.path.realpath(target_worktree)
-    found_registration = False
-    if wt_list_res.returncode == 0:
-        for block in wt_list_res.stdout.split("\n\n"):
-            lines = block.strip().splitlines()
-            worktree_line = next(
-                (l[9:].strip() for l in lines if l.startswith("worktree ")),
-                None,
-            )
-            if worktree_line and os.path.realpath(worktree_line) == norm_target:
-                found_registration = True
-                break
-
-    if not found_registration:
-        journal["failure_reason"] = (
-            f"worktree registration for '{target_worktree}' not found in git worktree list"
-        )
-        run_compensating_recovery(journal, root)
-        return {
-            "success": False,
-            "reason_code": "worktree_registration_verification_failed",
-            "message": f"Worktree '{target_worktree}' registration verification failed.",
-            "exact_next_action": "Check git worktree list integrity and retry.",
-        }
-
-    journal["phases"][PHASE_4_WORKTREE_CONFIRMED] = {
-        "status": "completed",
-        "worktree_path": target_worktree,
-    }
-    journal["phases"][PHASE_5_REGISTRATION_VERIFIED] = {
-        "status": "completed",
-        "registered": True,
-    }
-    journal["current_phase"] = PHASE_6_STATE_ESTABLISHED
-    save_phase_journal(journal)
-
-    # Phase 6: STATE_ESTABLISHED — Issue Lock Acquisition
-    from datetime import datetime, timezone
-    try:
-        lock_data = {
-            "remote": remote,
-            "org": org or "Scaled-Tech-Consulting",
-            "repo": repo or "Gitea-Tools",
-            "issue_number": issue_number,
-            "branch": target_branch,
-            "branch_name": target_branch,
-            "worktree_path": target_worktree,
-            "owner_session": owner_session or "prgs-author-95048-63667752",
-            "claimant": {
-                "username": active_identity or "jcwalker3",
-                "profile": active_profile or "prgs-author",
-            },
-            "assignment_id": assignment_id,
             "lease_id": lease_id,
-            "expected_base_sha": live_master_sha,
+            "assignment_id": assignment_id,
-            "created_at": datetime.now(timezone.utc).isoformat(),
+            "idempotency_key": key,
+            "lock_state": lock_res,
+            "phase_journal": journal,
+            "exact_next_action": (
+                f"Call gitea_whoami, then gitea_resolve_task_capability(task='work_issue', worktree_path='{target_worktree}') "
+                "and proceed with author implementation in the bootstrapped worktree."
+            ),
         }
-        lock_res = issue_lock_store.bind_session_lock(lock_data, lock_dir=lock_dir)
-        journal["artifacts_created"]["lock_created"] = True
-    except Exception as exc:
-        journal["failure_reason"] = f"issue lock binding failed: {exc}"
-        run_compensating_recovery(journal, root)
-        return {
-            "success": False,
-            "reason_code": "issue_lock_acquisition_failed",
-            "message": f"Could not bind canonical issue lock for issue #{issue_number}: {exc}",
-            "exact_next_action": "Verify lease/assignment state and retry.",
-        }
-
-    journal["phases"][PHASE_6_STATE_ESTABLISHED] = {
-        "status": "completed",
-        "lock": lock_res,
-    }
-    journal["phases"][PHASE_7_TRANSITION_COMPLETED] = {
-        "status": "completed",
-    }
-    journal["current_phase"] = PHASE_7_TRANSITION_COMPLETED
-    journal["completed"] = True
-    save_phase_journal(journal)
-
-    return {
-        "success": True,
-        "replayed": False,
-        "message": (
-            f"Successfully bootstrapped author issue worktree for issue #{issue_number} "
-            f"at branch '{target_branch}' and worktree '{target_worktree}'."
-        ),
-        "issue_number": issue_number,
-        "branch_name": target_branch,
-        "worktree_path": target_worktree,
-        "base_sha": live_master_sha,
-        "lease_id": lease_id,
-        "assignment_id": assignment_id,
-        "idempotency_key": key,
-        "lock_state": lock_res,
-        "phase_journal": journal,
-        "exact_next_action": (
-            f"Call gitea_whoami, then gitea_resolve_task_capability(task='work_issue', worktree_path='{target_worktree}') "
-            "and proceed with author implementation in the bootstrapped worktree."
-        ),
-    }

author_mutation_worktree.py

@@ -40,21 +40,41 @@ def _normalize_path(path: str) -> str:
     return (path or "").replace("\\", "/").rstrip("/")
 
 
-def is_path_under_branches(path: str, project_root: str | None = None) -> bool:
+def get_canonical_branches_root(project_root: str | None = None) -> str:
-    """True when *path* resolves inside ``<project_root>/branches/``."""
+    """Resolve the exact canonical branches root directory for the repository."""
-    normalized = _normalize_path(path)
-    if not normalized:
-        return False
-    if "/branches/" in f"{normalized}/":
-        return True
-    if normalized.endswith("/branches"):
-        return True
     if project_root:
-        root = _normalize_path(os.path.realpath(project_root))
+        root = os.path.realpath(project_root)
-        real = _normalize_path(os.path.realpath(path))
+    else:
-        if real.startswith(f"{root}/"):
+        root = os.path.realpath(os.getcwd())
-            rel = real[len(root) + 1 :]
+
-            return rel == "branches" or rel.startswith("branches/")
+    norm = root.replace("\\", "/")
+    if "/branches/" in norm:
+        base_part = norm.split("/branches/")[0]
+        return os.path.realpath(os.path.join(base_part, "branches"))
+    elif norm.endswith("/branches"):
+        return os.path.realpath(norm)
+
+    return os.path.realpath(os.path.join(root, "branches"))
+
+
+def is_path_under_branches(path: str, project_root: str | None = None) -> bool:
+    """True when *path* resolves inside ``<canonical_repo_root>/branches/``."""
+    if not path or not str(path).strip():
+        return False
+    try:
+        real_path = os.path.realpath(os.path.abspath(str(path).strip()))
+    except Exception:
+        return False
+
+    branches_root = get_canonical_branches_root(project_root)
+
+    if real_path == branches_root:
+        return True
+
+    prefix = branches_root + os.sep
+    if real_path.startswith(prefix):
+        return True
+
     return False
 
 

tests/test_author_issue_bootstrap.py

@@ -13,6 +13,18 @@ import author_issue_bootstrap
 import task_capability_map
 
 
+def _concurrent_bootstrap_worker(args: tuple[str, int, str, str, str, str]) -> dict:
+    repo_dir, issue_num, key, lock_dir, journal_dir, master_sha = args
+    os.environ["GITEA_BOOTSTRAP_JOURNAL_DIR"] = journal_dir
+    return author_issue_bootstrap.bootstrap_author_issue_worktree(
+        issue_number=issue_num,
+        canonical_repo_root=repo_dir,
+        expected_base_sha=master_sha,
+        idempotency_key=key,
+        lock_dir=lock_dir,
+    )
+
+
 class TestAuthorIssueBootstrap(unittest.TestCase):
     """Test suite covering AC1-AC10 and comment #14959 specification."""
 
@@ -187,6 +199,149 @@ class TestAuthorIssueBootstrap(unittest.TestCase):
         branch_check = subprocess.run(["git", "-C", self.repo_dir, "rev-parse", "--verify", journal["branch_name"]], capture_output=True, text=True, check=False)
         self.assertNotEqual(branch_check.returncode, 0)
 
+    def test_cross_process_concurrency(self):
+        """Review #525 Finding 1: Genuine cross-process concurrency locking prevents corruption."""
+        import concurrent.futures
+
+        key = "test_concurrent_key_850"
+        args = (self.repo_dir, 850, key, self.lock_dir, self.journal_dir, self.master_sha)
+
+        with concurrent.futures.ProcessPoolExecutor(max_workers=2) as executor:
+            fut1 = executor.submit(_concurrent_bootstrap_worker, args)
+            fut2 = executor.submit(_concurrent_bootstrap_worker, args)
+            res1 = fut1.result(timeout=10)
+            res2 = fut2.result(timeout=10)
+
+        self.assertTrue(res1["success"], f"res1 failed: {res1}")
+        self.assertTrue(res2["success"], f"res2 failed: {res2}")
+        # One process performs creation, the other process receives idempotent replay
+        replayed_count = sum(1 for r in (res1, res2) if r.get("replayed"))
+        created_count = sum(1 for r in (res1, res2) if not r.get("replayed"))
+        self.assertEqual(replayed_count, 1)
+        self.assertEqual(created_count, 1)
+        self.assertEqual(res1["worktree_path"], res2["worktree_path"])
+
+    def test_interrupted_replay_preserves_artifacts_created_provenance(self):
+        """Review #525 Finding 2: Replaying incomplete journal preserves creation provenance monotonically."""
+        key = "test_key_interrupted_replay_1"
+        branch = "fix/issue-850-interrupted-replay"
+        wt_path = os.path.join(self.branches_dir, "fix-issue-850-interrupted-replay")
+
+        # Simulate Phase 2/3 completion where branch and worktree directory were created by this transition
+        journal = {
+            "idempotency_key": key,
+            "issue_number": 850,
+            "branch_name": branch,
+            "worktree_path": wt_path,
+            "active_identity": "jcwalker3",
+            "active_profile": "prgs-author",
+            "remote": "prgs",
+            "org": "Scaled-Tech-Consulting",
+            "repo": "Gitea-Tools",
+            "phases": {
+                author_issue_bootstrap.PHASE_1_REQUEST_ACCEPTED: {"status": "completed"},
+                author_issue_bootstrap.PHASE_2_BRANCH_CONFIRMED: {"status": "completed", "created": True},
+            },
+            "artifacts_created": {
+                "branch_created": True,
+                "worktree_dir_created": True,
+                "worktree_registered": True,
+                "lock_created": False,
+            },
+            "current_phase": author_issue_bootstrap.PHASE_3_PATH_RESERVED,
+            "completed": False,
+        }
+        # Pre-create the branch and worktree on disk to simulate partial state after crash
+        subprocess.run(["git", "-C", self.repo_dir, "branch", branch, self.master_sha], check=True, capture_output=True)
+        subprocess.run(["git", "-C", self.repo_dir, "worktree", "add", wt_path, branch], check=True, capture_output=True)
+        author_issue_bootstrap.save_phase_journal(journal)
+
+        # Now resume/replay the transition but simulate lock binding failure during Phase 6
+        with unittest.mock.patch("issue_lock_store.bind_session_lock", side_effect=RuntimeError("Lock failure test")):
+            res = author_issue_bootstrap.bootstrap_author_issue_worktree(
+                issue_number=850,
+                canonical_repo_root=self.repo_dir,
+                branch_name=branch,
+                worktree_path=wt_path,
+                idempotency_key=key,
+                lock_dir=self.lock_dir,
+            )
+
+        self.assertFalse(res["success"])
+        self.assertEqual(res.get("reason_code"), "issue_lock_acquisition_failed")
+
+        # Verify that compensating recovery correctly deleted transition-created branch & worktree
+        # because creation provenance was preserved across replay (NOT downgraded to False!)
+        self.assertFalse(os.path.exists(wt_path))
+        branch_check = subprocess.run(["git", "-C", self.repo_dir, "rev-parse", "--verify", branch], capture_output=True, text=True, check=False)
+        self.assertNotEqual(branch_check.returncode, 0)
+
+    def test_transition_created_only_compensation(self):
+        """Review #525 Finding 4: Preexisting branch is NOT deleted by compensation when only worktree was transition-created."""
+        key = "test_key_preexisting_branch_compensation"
+        preexisting_branch = "fix/issue-850-preexisting"
+        wt_path = os.path.join(self.branches_dir, "fix-issue-850-preexisting")
+
+        # Create branch BEFORE bootstrap (preexisting branch)
+        subprocess.run(["git", "-C", self.repo_dir, "branch", preexisting_branch, self.master_sha], check=True, capture_output=True)
+
+        # Call bootstrap with simulated failure during Phase 6 (lock binding)
+        with unittest.mock.patch("issue_lock_store.bind_session_lock", side_effect=RuntimeError("Simulated lock failure")):
+            res = author_issue_bootstrap.bootstrap_author_issue_worktree(
+                issue_number=850,
+                canonical_repo_root=self.repo_dir,
+                branch_name=preexisting_branch,
+                worktree_path=wt_path,
+                idempotency_key=key,
+                lock_dir=self.lock_dir,
+            )
+
+        self.assertFalse(res["success"])
+        # Worktree dir was created by transition -> removed by compensation
+        self.assertFalse(os.path.exists(wt_path))
+
+        # Preexisting branch was NOT created by transition -> MUST BE PRESERVED!
+        branch_check = subprocess.run(["git", "-C", self.repo_dir, "rev-parse", "--verify", preexisting_branch], capture_output=True, text=True, check=False)
+        self.assertEqual(branch_check.returncode, 0, "Preexisting branch was deleted by mistake!")
+
+    def test_incompatible_idempotency_replay_refusal(self):
+        """Review #525 Finding 4: Replaying key with incompatible parameters returns refusal."""
+        key = "test_key_incompatible_replay"
+        res1 = author_issue_bootstrap.bootstrap_author_issue_worktree(
+            issue_number=850,
+            canonical_repo_root=self.repo_dir,
+            branch_name="fix/issue-850-param-a",
+            idempotency_key=key,
+            lock_dir=self.lock_dir,
+        )
+        self.assertTrue(res1["success"])
+
+        # Second call with different branch_name
+        res2 = author_issue_bootstrap.bootstrap_author_issue_worktree(
+            issue_number=850,
+            canonical_repo_root=self.repo_dir,
+            branch_name="fix/issue-850-param-b",
+            idempotency_key=key,
+            lock_dir=self.lock_dir,
+        )
+        self.assertFalse(res2["success"])
+        self.assertEqual(res2.get("reason_code"), "incompatible_idempotency_replay")
+
+    def test_exact_next_action_satisfiable_via_mcp(self):
+        """Review #525 Finding 4: exact_next_action provides satisfiable MCP actions, not shell commands."""
+        key = "test_key_next_action_mcp"
+        stale_sha = "0000000000000000000000000000000000000000"
+        res = author_issue_bootstrap.bootstrap_author_issue_worktree(
+            issue_number=850,
+            canonical_repo_root=self.repo_dir,
+            expected_base_sha=stale_sha,
+            lock_dir=self.lock_dir,
+        )
+        next_action = res.get("exact_next_action", "")
+        self.assertNotIn("scripts/worktree-start", next_action)
+        self.assertNotIn("git worktree add", next_action)
+        self.assertNotIn("bash", next_action.lower())
+
     def test_task_capability_map_integration(self):
         """Verify task_capability_map has bootstrap_author_issue_worktree configured correctly."""
         self.assertEqual(task_capability_map.required_role("bootstrap_author_issue_worktree"), "author")

tests/test_author_mutation_worktree.py

@@ -28,6 +28,21 @@ class TestPathUnderBranches(unittest.TestCase):
             amw.is_path_under_branches("/repo/other-checkout", self.ROOT)
         )
 
+    def test_unrelated_branches_dir_fails(self):
+        self.assertFalse(
+            amw.is_path_under_branches("/tmp/branches/evil", self.ROOT)
+        )
+
+    def test_prefix_confusion_fails(self):
+        self.assertFalse(
+            amw.is_path_under_branches(f"{self.ROOT}/branches-other/foo", self.ROOT)
+        )
+
+    def test_traversal_fails(self):
+        self.assertFalse(
+            amw.is_path_under_branches(f"{self.ROOT}/branches/../evil", self.ROOT)
+        )
+
 
 class TestAssessAuthorMutationWorktree(unittest.TestCase):
     ROOT = "/repo/Gitea-Tools"