dirty_orphan_worktree_recovery.py

@@ -725,13 +725,14 @@ def prepare_recovery_worktree(
         )
         head = (probe.stdout or "").strip()
         if probe.returncode != 0 or head != remote_head:
-            # Allow dirty recovery worktree after prior partial apply.
             return {
-                "success": True,
+                "success": False,
                 "created": False,
                 "resumed": True,
                 "head": head,
-                "reasons": [],
+                "reasons": [
+                    f"resume recovery worktree HEAD '{head}' does not match expected remote HEAD '{remote_head}'"
+                ],
             }
         return {
             "success": True,
@@ -741,7 +742,8 @@ def prepare_recovery_worktree(
             "reasons": [],
         }
 
-    # Create detached-at-head worktree then force branch association carefully.
+    # Create detached-at-head worktree. Do not run git checkout -B because
+    # the source worktree holds the branch name (Git exit 128).
     add = git.run(
         [
             "git",
@@ -761,19 +763,6 @@ def prepare_recovery_worktree(
                 f"git worktree add failed: {(add.stderr or add.stdout or '').strip()}"
             ],
         }
-    # Create/update local branch pointer without moving source.
-    br = git.run(
-        ["git", "checkout", "-B", branch_name],
-        cwd=recovery_worktree_path,
-    )
-    if br.returncode != 0:
-        return {
-            "success": False,
-            "created": True,
-            "reasons": [
-                f"branch checkout failed: {(br.stderr or br.stdout or '').strip()}"
-            ],
-        }
     return {
         "success": True,
         "created": True,
@@ -965,6 +954,27 @@ def run_dirty_orphan_recovery(
     journal["source_still_present"] = os.path.isdir(source_worktree_path)
     save_journal(journal, journal_dir=journal_dir)
 
+    # #860 F4: Do NOT finalize session binding if conflicts remain.
+    if conflicts:
+        return {
+            "success": False,
+            "performed": True,
+            "outcome": CONFLICTS_PRESENT,
+            "reasons": [
+                "conflicts present: manual resolution required before session binding (fail closed)"
+            ],
+            "conflicts": list(conflicts),
+            "recovery_worktree_path": recovery_worktree_path,
+            "source_worktree_path": source_worktree_path,
+            "source_frozen": True,
+            "journal": journal,
+            "evidence": {
+                "written_paths": written,
+                "conflict_state_path": conflict_state_path,
+                "accepted_head": expected_remote_head,
+            },
+        }
+
     # Phase 5: bind session (optional for pure assessor tests).
     pid = session_pid if session_pid is not None else os.getpid()
     lock_record = build_recovery_lock_record(
@@ -992,7 +1002,9 @@ def run_dirty_orphan_recovery(
             except Exception:
                 prior_gen = None
             _ils.bind_session_lock(
-                lock_record, expected_generation=prior_gen
+                lock_record,
+                expected_generation=prior_gen,
+                recovery_sanctioned=True,
             )
         else:
             lock_writer(lock_record)
@@ -1013,13 +1025,12 @@ def run_dirty_orphan_recovery(
     journal["complete"] = True
     save_journal(journal, journal_dir=journal_dir)
 
-    outcome = CONFLICTS_PRESENT if conflicts else RECOVERY_COMPLETED
     return {
         "success": True,
         "performed": True,
-        "outcome": outcome,
+        "outcome": RECOVERY_COMPLETED,
         "reasons": [],
-        "conflicts": list(conflicts),
+        "conflicts": [],
         "recovery_worktree_path": recovery_worktree_path,
         "source_worktree_path": source_worktree_path,
         "source_frozen": True,

gitea_mcp_server.py

@@ -1450,7 +1450,7 @@ def verify_preflight_purity(
             dirty_files = sorted(
                 _parse_porcelain_entries(_get_workspace_porcelain(workspace))
             )
-            if dirty_files:
+            if dirty_files and task != "commit_files":
                 raise RuntimeError(
                     nwb.format_namespace_workspace_binding_error(
                         role_kind=role,
@@ -4460,8 +4460,6 @@ def gitea_recover_dirty_orphaned_issue_worktree(
             observed_remote = (probe.stdout or "").strip().split()[0]
     except Exception:
         observed_remote = ""
-    if not observed_remote:
-        observed_remote = (expected_remote_head or "").strip()
 
     registered = False
     try:
@@ -4482,6 +4480,41 @@ def gitea_recover_dirty_orphaned_issue_worktree(
         src, project_root
     )
 
+    competing_locks: list[dict] = []
+    try:
+        all_live = issue_lock_store.list_live_locks()
+        for l in all_live:
+            if l.get("issue_number") == issue_number:
+                wt = l.get("worktree_path")
+                if not wt or not issue_lock_store._same_realpath(wt, src):
+                    competing_locks.append(l)
+    except Exception:
+        competing_locks = []
+
+    wf_active = False
+    wf_expired = True
+    try:
+        db, _ = _control_plane_db_or_error()
+        if db is not None:
+            active_leases_data = lease_lifecycle.list_active_leases(
+                db,
+                remote=remote if remote in REMOTES else remote,
+                org=o,
+                repo=r,
+            )
+            leases_list = active_leases_data.get("leases") or []
+            for l in leases_list:
+                if l.get("work_number") == issue_number and l.get("work_kind") == "issue":
+                    fresh = l.get("freshness") or {}
+                    if fresh.get("status") == "active":
+                        wf_active = True
+                        wf_expired = False
+                    elif fresh.get("status") in ("expired", "stale_dead_process"):
+                        wf_active = False
+                        wf_expired = True
+    except Exception:
+        pass
+
     assessment = dirty_orphan_worktree_recovery.assess_dirty_orphan_recovery(
         existing_lock,
         issue_number=issue_number,
@@ -4500,10 +4533,10 @@ def gitea_recover_dirty_orphaned_issue_worktree(
         observed_local_head=observed_local,
         observed_remote_head=observed_remote,
         observed_dirty_fingerprints=observed_fps,
-        competing_live_locks=[],
+        competing_live_locks=competing_locks,
         competing_live_sessions=[],
-        workflow_lease_active=False,
+        workflow_lease_active=wf_active,
-        workflow_lease_expired=True,
+        workflow_lease_expired=wf_expired,
         canonical_repo_root=canonical_root,
         worktree_registered=registered,
         current_pid=os.getpid(),

issue_lock_provenance.py

@@ -16,11 +16,13 @@ ISSUE_LOCK_FILE = os.environ.get("GITEA_ISSUE_LOCK_FILE", "/tmp/gitea_issue_lock
 SOURCE_LOCK_ISSUE = "gitea_lock_issue"
 SOURCE_LOCK_ADOPTION = "gitea_lock_issue_adoption"
 SOURCE_OPERATOR_OVERRIDE = "operator_override"
+SOURCE_RECOVER_DIRTY_ORPHANED = "gitea_recover_dirty_orphaned_issue_worktree"
 
 SANCTIONED_LOCK_SOURCES = frozenset({
     SOURCE_LOCK_ISSUE,
     SOURCE_LOCK_ADOPTION,
     SOURCE_OPERATOR_OVERRIDE,
+    SOURCE_RECOVER_DIRTY_ORPHANED,
 })
 
 _OPERATOR_OVERRIDE_ENV = "GITEA_ISSUE_LOCK_OPERATOR_OVERRIDE"

issue_lock_store.py

@@ -169,6 +169,7 @@ def bind_session_lock(
     *,
     expected_generation: int | None = None,
     renewal_sanctioned: bool = False,
+    recovery_sanctioned: bool = False,
 ) -> str:
     """Persist a keyed lock and bind it to the current process session.
 
@@ -213,7 +214,9 @@ def bind_session_lock(
     try:
         with _exclusive_file_lock(sentinel):
             existing = read_lock_file(path)
-            overwrite_block = assess_foreign_lock_overwrite(existing, record)
+            overwrite_block = assess_foreign_lock_overwrite(
+                existing, record, recovery_sanctioned=recovery_sanctioned
+            )
             if overwrite_block:
                 raise RuntimeError(overwrite_block)
             lease_block = assess_same_issue_lease_conflict(
@@ -222,6 +225,7 @@ def bind_session_lock(
                 branch_name=str(record.get("branch_name") or ""),
                 worktree_path=str(record.get("worktree_path") or ""),
                 renewal_sanctioned=renewal_sanctioned,
+                recovery_sanctioned=recovery_sanctioned,
             )
             if lease_block:
                 raise RuntimeError(lease_block)
@@ -517,6 +521,7 @@ def assess_same_issue_lease_conflict(
     worktree_path: str,
     operation_type: str = AUTHOR_ISSUE_WORK_LEASE,
     renewal_sanctioned: bool = False,
+    recovery_sanctioned: bool = False,
     now: datetime | None = None,
 ) -> str | None:
     """Return a fail-closed error when a competing live lease blocks acquisition.
@@ -548,6 +553,8 @@ def assess_same_issue_lease_conflict(
         existing_branch == branch_name
         and _same_realpath(str(existing_worktree or ""), worktree_path)
     )
+    if recovery_sanctioned and existing_issue == issue_number and existing_branch == branch_name:
+        return None
     if is_lease_expired(existing_lock, now=now):
         # #760 AC1/AC2: exact-owner renewal is a different disposition from
         # foreign takeover and is evaluated first. Before this, both branches
@@ -578,10 +585,26 @@ def assess_same_issue_lease_conflict(
     )
 
 
+def _lock_claimant(lock: dict[str, Any] | None) -> dict[str, str]:
+    if not isinstance(lock, dict):
+        return {}
+    claimant = lock.get("claimant")
+    if not isinstance(claimant, dict):
+        lease = lock.get("work_lease")
+        claimant = lease.get("claimant") if isinstance(lease, dict) else None
+    if not isinstance(claimant, dict):
+        return {}
+    return {
+        "username": str(claimant.get("username") or ""),
+        "profile": str(claimant.get("profile") or ""),
+    }
+
+
 def assess_foreign_lock_overwrite(
     existing_lock: dict[str, Any] | None,
     incoming_lock: dict[str, Any],
     *,
+    recovery_sanctioned: bool = False,
     now: datetime | None = None,
 ) -> str | None:
     """Block writes that would clobber an unrelated live lease on the same key."""
@@ -596,8 +619,31 @@ def assess_foreign_lock_overwrite(
     )
     if same_issue and same_branch and same_worktree:
         return None
-    if not is_lease_live(existing_lock, now=now):
+
+    existing_claimant = _lock_claimant(existing_lock)
+    incoming_claimant = _lock_claimant(incoming_lock)
+    same_claimant = (
+        bool(existing_claimant.get("username"))
+        and existing_claimant.get("username") == incoming_claimant.get("username")
+        and existing_claimant.get("profile") == incoming_claimant.get("profile")
+    )
+
+    if recovery_sanctioned and same_issue and same_branch and same_claimant:
         return None
+
+    if not is_lease_live(existing_lock, now=now):
+        # #860 F8: A non-live or PID-less lock still blocks foreign overwrite
+        # unless same claimant or sanctioned reclaim is proven.
+        if not same_claimant and same_issue:
+            reclaim = assess_expired_lock_reclaim(existing_lock, now=now)
+            if not reclaim.get("reclaim_allowed"):
+                return (
+                    "Refusing foreign overwrite of non-live issue lock "
+                    f"(issue #{existing_lock.get('issue_number')}, owner '{existing_claimant.get('username')}') "
+                    "without sanctioned reclaim proof (fail closed)"
+                )
+        return None
+
     return (
         "Refusing to overwrite a live foreign issue lock "
         f"(issue #{existing_lock.get('issue_number')}, "

scripts/worktree-start

@@ -43,19 +43,21 @@ repo_root="$(cd "$script_dir/.." && pwd)"
 
 # Enforce issue-linked, traceable branch names (issue → branch → worktree → PR).
 if [[ "$allow_unlinked" -eq 0 ]]; then
-  locked_branch=$(python3 -c "
+  if [[ "$dry_run" -eq 0 ]] && [[ ! "$branch" =~ ^review/pr-[0-9]+-.+ ]]; then
+    locked_branch=$(python3 -c "
 import sys
 sys.path.insert(0, '$repo_root')
 import issue_lock_store
 print(issue_lock_store.resolve_locked_branch_for_session('$branch'))
 ")
-  if [[ -z "$locked_branch" ]]; then
+    if [[ -z "$locked_branch" ]]; then
-    echo "Error: No session issue lock is bound. Call gitea_lock_issue before branch creation (fail closed)." >&2
+      echo "Error: No session issue lock is bound. Call gitea_lock_issue before branch creation (fail closed)." >&2
-    exit 2
+      exit 2
-  fi
+    fi
-  if [[ "$branch" != "$locked_branch" ]]; then
+    if [[ "$branch" != "$locked_branch" ]]; then
-    echo "Error: Requested branch '$branch' does not match locked branch '$locked_branch' (fail closed)." >&2
+      echo "Error: Requested branch '$branch' does not match locked branch '$locked_branch' (fail closed)." >&2
-    exit 2
+      exit 2
+    fi
   fi
 
   if [[ "$branch" =~ ^(fix|feat|docs|chore)/issue-[0-9]+-.+ ]] \

task_capability_map.py

@@ -486,6 +486,11 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
 # merger lease (#763).
 _PREFLIGHT_TASK_TRANSITIONS = frozenset({
     ("review_pr", "acquire_reviewer_pr_lease"),
+    ("work_issue", "lock_issue"),
+    ("work_issue", "recover_dirty_orphaned_issue_worktree"),
+    ("work_issue", "gitea_recover_dirty_orphaned_issue_worktree"),
+    ("work_issue", "commit_files"),
+    ("work_issue", "gitea_commit_files"),
 })
 
 

tests/test_dirty_orphan_worktree_recovery.py

@@ -305,7 +305,8 @@ class CrashSafeRecovery(unittest.TestCase):
             local_head_contents={"c.py": b"local-base"},
             remote_head_contents={"c.py": b"remote-changed"},
         )
-        self.assertTrue(result["success"])
+        # #860 F4: session binding is NOT finalized while conflicts remain
+        self.assertFalse(result["success"])
         self.assertEqual(result["outcome"], dorec.CONFLICTS_PRESENT)
         sidecar = os.path.join(self.recovery, "c.py.recovered-dirty")
         self.assertTrue(os.path.isfile(sidecar))
@@ -403,10 +404,8 @@ class JournalSymlinkRefusal(unittest.TestCase):
                 fh.write("{}")
             link = os.path.join(tmp, "link.json")
             os.symlink(real, link)
-            # Point journal path helper via env
             key = "symlink-test"
             jdir = tmp
-            # Craft path that is a symlink by saving then replacing
             path = dorec._journal_path(key, journal_dir=jdir)
             with open(path, "w", encoding="utf-8") as fh:
                 json.dump({"idempotency_key": key}, fh)
@@ -418,5 +417,67 @@ class JournalSymlinkRefusal(unittest.TestCase):
             shutil.rmtree(tmp, ignore_errors=True)
 
 
+class RealGitMultiWorktreeIntegration(unittest.TestCase):
+    def setUp(self):
+        import subprocess
+        self.tmp = tempfile.mkdtemp(prefix="git-integration-")
+        self.repo = os.path.join(self.tmp, "repo")
+        os.makedirs(self.repo, exist_ok=True)
+        subprocess.run(["git", "init"], cwd=self.repo, check=True, capture_output=True)
+        subprocess.run(["git", "config", "user.name", "Test User"], cwd=self.repo, check=True)
+        subprocess.run(["git", "config", "user.email", "[email protected]"], cwd=self.repo, check=True)
+        with open(os.path.join(self.repo, "init.txt"), "w") as fh:
+            fh.write("init")
+        subprocess.run(["git", "add", "."], cwd=self.repo, check=True)
+        subprocess.run(["git", "commit", "-m", "init"], cwd=self.repo, check=True)
+        branch = "fix/issue-999-test"
+        subprocess.run(["git", "branch", branch], cwd=self.repo, check=True)
+        self.branches = os.path.join(self.repo, "branches")
+        self.source = os.path.join(self.branches, "issue-999-test")
+        subprocess.run(["git", "worktree", "add", self.source, branch], cwd=self.repo, check=True)
+        self.dirty_path = os.path.join(self.source, "dirty.txt")
+        with open(self.dirty_path, "w") as fh:
+            fh.write("dirty-data")
+
+    def tearDown(self):
+        shutil.rmtree(self.tmp, ignore_errors=True)
+
+    def test_prepare_recovery_worktree_detached_no_exit_128(self):
+        import subprocess
+        head_sha = subprocess.check_output(["git", "rev-parse", "HEAD"], cwd=self.repo, text=True).strip()
+        rec_wt = os.path.join(self.branches, "recovery-issue-999-test")
+        res = dorec.prepare_recovery_worktree(
+            canonical_repo_root=self.repo,
+            recovery_worktree_path=rec_wt,
+            branch_name="fix/issue-999-test",
+            remote_head=head_sha,
+        )
+        self.assertTrue(res["success"], res.get("reasons"))
+        self.assertTrue(os.path.isdir(rec_wt))
+
+    def test_real_lock_rebind_recovery_sanctioned(self):
+        lock_dir = os.path.join(self.tmp, "locks")
+        rec_wt = os.path.join(self.branches, "recovery-issue-999-test")
+        os.makedirs(rec_wt, exist_ok=True)
+        record = {
+            "remote": "prgs",
+            "org": "Example-Org",
+            "repo": "Example-Repo",
+            "issue_number": 999,
+            "branch_name": "fix/issue-999-test",
+            "worktree_path": rec_wt,
+            "claimant": {"username": "author-user", "profile": "prgs-author"},
+        }
+        record_src = dict(record)
+        record_src["worktree_path"] = self.source
+        issue_lock_store.bind_session_lock(record_src, lock_dir=lock_dir)
+        path = issue_lock_store.bind_session_lock(
+            record,
+            lock_dir=lock_dir,
+            recovery_sanctioned=True,
+        )
+        self.assertTrue(os.path.isfile(path))
+
+
 if __name__ == "__main__":
     unittest.main()