fix(reconcile): safely resolve worktree bindings whose paths are missing (Closes #970)

control_plane_db.py

@@ -1913,6 +1913,88 @@ class ControlPlaneDB:
                 ),
             )
 
+    def retire_lease_worktree_path(
+        self,
+        lease_id: str,
+        *,
+        expected_path: str | None = None,
+        reason: str = "missing_worktree_path_retired",
+    ) -> dict[str, Any]:
+        """Retire a missing worktree_path binding from a control-plane lease (#970).
+
+        Clears worktree_path on the lease row, updates provenance_json with
+        durable retirement audit proof, and writes a worktree_binding_retired event.
+        Fail closed: if expected_path is provided and does not match current
+        worktree_path, the update is refused to prevent racing mutations.
+        """
+        now_s = _ts()
+        with self._tx() as conn:
+            cols = self._lease_columns(conn)
+            row = conn.execute(
+                "SELECT * FROM leases WHERE lease_id = ?",
+                (lease_id,),
+            ).fetchone()
+            if not row:
+                raise ControlPlaneError(f"unknown lease_id {lease_id}")
+
+            current_wt = (dict(row).get("worktree_path") or "").strip()
+            if expected_path and current_wt and os.path.realpath(current_wt) != os.path.realpath(expected_path):
+                raise ControlPlaneError(
+                    f"cannot retire lease {lease_id} worktree_path: expected '{expected_path}' "
+                    f"does not match current '{current_wt}' (fail closed)"
+                )
+
+            # Parse and update provenance_json
+            raw_prov = dict(row).get("provenance_json") or "{}"
+            try:
+                prov = json.loads(raw_prov) if isinstance(raw_prov, str) else dict(raw_prov)
+            except Exception:
+                prov = {}
+            if not isinstance(prov, dict):
+                prov = {}
+
+            prior_path = current_wt or prov.get("worktree_path")
+            prov.update({
+                "worktree_path_retired": True,
+                "retired_worktree_path": prior_path,
+                "retired_at": now_s,
+                "retirement_reason": reason,
+                "worktree_path": "",
+            })
+            prov_json = json.dumps(prov)
+
+            if "worktree_path" in cols:
+                conn.execute(
+                    "UPDATE leases SET worktree_path = '', provenance_json = ? WHERE lease_id = ?",
+                    (prov_json, lease_id),
+                )
+            else:
+                conn.execute(
+                    "UPDATE leases SET provenance_json = ? WHERE lease_id = ?",
+                    (prov_json, lease_id),
+                )
+
+            conn.execute(
+                """
+                INSERT INTO events(work_item_id, event_type, message, created_at)
+                VALUES (?, 'worktree_binding_retired', ?, ?)
+                """,
+                (
+                    row["work_item_id"],
+                    f"lease {lease_id} worktree_path '{prior_path}' retired: {reason}",
+                    now_s,
+                ),
+            )
+
+            return {
+                "lease_id": lease_id,
+                "retired": True,
+                "prior_worktree_path": prior_path,
+                "retired_at": now_s,
+                "reason": reason,
+            }
+
+
     def abandon_lease(
         self,
         *,
@@ -3051,3 +3133,32 @@ class ControlPlaneDB:
             "live_lease_id": None if live_lease_id is None else str(live_lease_id),
             "reconcile_action": "reconcile_required" if stale else "safe_to_resume",
         }
+
+    def retire_session_checkpoint_worktree_path(
+        self,
+        session_id: str,
+        *,
+        checkpoint_id: str | None = None,
+        expected_path: str | None = None,
+        reason: str = "missing_worktree_path_retired",
+    ) -> dict[str, Any]:
+        """Retire a missing worktree_path from session_checkpoints (#970)."""
+        now_s = _ts()
+        with self._tx() as conn:
+            if checkpoint_id:
+                conn.execute(
+                    "UPDATE session_checkpoints SET worktree_path = '', updated_at = ? WHERE checkpoint_id = ?",
+                    (now_s, checkpoint_id),
+                )
+            elif session_id:
+                conn.execute(
+                    "UPDATE session_checkpoints SET worktree_path = '', updated_at = ? WHERE session_id = ?",
+                    (now_s, session_id),
+                )
+            return {
+                "session_id": session_id,
+                "checkpoint_id": checkpoint_id,
+                "retired": True,
+                "retired_at": now_s,
+                "reason": reason,
+            }

docs/mcp-tool-inventory.md

@@ -65,6 +65,7 @@ that gates each call, not which tools exist.
 - `gitea_assess_work_issue_duplicate`
 - `gitea_assess_worktree_cleanup_integrity`
 - `gitea_audit_config`
+- `gitea_audit_missing_worktree_bindings`
 - `gitea_audit_runtime_recovery_contamination`
 - `gitea_audit_stable_branch_contamination`
 - `gitea_audit_worktree_cleanup`
@@ -126,16 +127,20 @@ that gates each call, not which tools exist.
 - `gitea_post_heartbeat`
 - `gitea_publish_unpublished_issue_branch`
 - `gitea_quarantine_contaminated_review`
+- `gitea_rebind_dirty_same_claimant_author_session`
 - `gitea_reclaim_expired_workflow_lease`
+- `gitea_reconcile_after_restart`
 - `gitea_reconcile_already_landed_pr`
 - `gitea_reconcile_issue_claims`
 - `gitea_reconcile_merged_cleanups`
+- `gitea_reconcile_missing_worktree_bindings`
 - `gitea_reconcile_superseded_by_merged_pr`
 - `gitea_record_daemon_process_kill_attempt`
 - `gitea_record_irrecoverable_decision_lock_provenance`
 - `gitea_record_pre_review_command`
 - `gitea_record_shell_spawn_outcome`
 - `gitea_record_stable_branch_push_attempt`
+- `gitea_recover_dirty_orphaned_issue_worktree`
 - `gitea_recover_incomplete_bootstrap_lock`
 - `gitea_release_merger_pr_lease`
 - `gitea_release_reviewer_pr_lease`

gitea_mcp_server.py

@@ -13547,6 +13547,110 @@ def gitea_scan_already_landed_open_prs(
     }
 
 
+@mcp.tool()
+def gitea_audit_missing_worktree_bindings(
+    remote: str = "dadeschools",
+    host: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+) -> dict:
+    """Read-only: audit and classify worktree bindings whose paths are missing on disk (#970).
+
+    Correlates missing-path bindings from control-plane leases, session checkpoints,
+    and issue locks with repository, host, branch, issue/PR, session, and lease state.
+    Distinguishes deleted worktrees from moved paths, host/mount failures, and live
+    leases/sessions.
+
+    Args:
+        remote: Known instance — 'dadeschools' or 'prgs'.
+        host: Override the Gitea host.
+        org: Override the owner/organization.
+        repo: Override the repository name.
+
+    Returns:
+        dict with audit counts, missing binding classifications, and resolution status.
+    """
+    read_block = _profile_operation_gate("gitea.read")
+    if read_block:
+        return {
+            "success": False,
+            "performed": False,
+            "reasons": read_block,
+            "permission_report": _permission_block_report("gitea.read"),
+        }
+
+    import missing_worktree_reconcile
+    db, _ = _control_plane_db_or_error()
+    root = _canonical_local_git_root()
+    h, o, r = _resolve(remote, host, org, repo)
+
+    return missing_worktree_reconcile.audit_missing_worktree_bindings(
+        db,
+        project_root=root,
+        remote=remote,
+        org=o,
+        repo=r,
+        host=h,
+    )
+
+
+@mcp.tool()
+def gitea_reconcile_missing_worktree_bindings(
+    dry_run: bool = True,
+    operator_authorized: bool = False,
+    remote: str = "dadeschools",
+    host: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+) -> dict:
+    """Audit and safely resolve (retire) missing worktree path bindings (#970).
+
+    Identifies confirmed stale deleted worktree bindings, re-validates them
+    immediately before mutation to prevent recreation races, and retires only the
+    exact stale bindings while preserving unrelated worktrees and Git metadata.
+
+    Args:
+        dry_run: If True (default), reports planned mutations without modifying state.
+        operator_authorized: Explicit operator authorization required for apply mode.
+        remote: Known instance — 'dadeschools' or 'prgs'.
+        host: Override the Gitea host.
+        org: Override the owner/organization.
+        repo: Override the repository name.
+
+    Returns:
+        dict with before/after audit state, resolutions, and dimension status.
+    """
+    read_block = _profile_operation_gate("gitea.read")
+    if read_block:
+        return {
+            "success": False,
+            "performed": False,
+            "reasons": read_block,
+            "permission_report": _permission_block_report("gitea.read"),
+        }
+
+    import missing_worktree_reconcile
+    db, db_errs = _control_plane_db_or_error()
+    root = _canonical_local_git_root()
+    h, o, r = _resolve(remote, host, org, repo)
+
+    profile = get_profile()
+    role = profile.get("role") or ""
+    workflow_authorized = role in ("reconciler", "author", "controller")
+
+    return missing_worktree_reconcile.reconcile_missing_worktree_bindings(
+        db,
+        project_root=root,
+        remote=remote,
+        org=o,
+        repo=r,
+        host=h,
+        dry_run=dry_run,
+        operator_authorized=operator_authorized,
+        workflow_authorized=workflow_authorized,
+    )
+
+
 @mcp.tool()
 def gitea_audit_worktree_cleanup(
     remote: str = "dadeschools",

missing_worktree_reconcile.py

@@ -0,0 +1,611 @@
+"""Sanctioned reconciliation for worktree bindings whose paths are missing (#970).
+
+When control-plane leases, session checkpoints, or issue locks hold a
+``worktree_path`` pointing to a filesystem path that no longer exists on disk,
+this module provides a fail-closed, auditable classification and resolution
+workflow.
+
+Core principles:
+1. **Multi-dimensional Correlation**:
+   Correlates the binding with repository, host, branch, issue/PR, session,
+   and lease state.
+2. **Safety-first Distinction**:
+   - Distinguishes a deleted worktree from a moved path, unavailable host/mount,
+     live lease, active session, or temporary filesystem problem.
+   - Host / mount failure (e.g. repo root or branches/ directory inaccessible)
+     blocks retirement (``unavailable_host_or_mount``).
+   - Moved worktree (found under another path in ``git worktree list``)
+     blocks retirement and reports the new location (``moved_worktree_path``).
+   - Live lease or active session/process blocks retirement (``live_lease_protected``,
+     ``live_session_protected``).
+3. **Atomic Pre-Mutation Re-Validation**:
+   Immediately before committing a mutation, re-evaluates filesystem existence,
+   git worktree list, host health, and lease status to prevent recreation races.
+4. **Targeted Retirement**:
+   Retires only the exact confirmed stale binding (clearing ``worktree_path`` on
+   the lease/checkpoint/lock row and recording durable provenance) while
+   preserving unrelated worktrees, leases, branches, and git metadata.
+5. **Dry-Run & Operator Authorization**:
+   Supports dry-run inspection and requires explicit cleanup authorization or
+   reconciler workflow for mutation.
+6. **Idempotency & Durable Proof**:
+   Repeated execution on an already-retired binding is safe, idempotent, and
+   produces durable audit evidence.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import subprocess
+from typing import Any, Mapping, Sequence
+
+import control_plane_db as cpd
+import issue_lock_store
+import lease_lifecycle
+from merged_cleanup_reconcile import list_local_worktrees
+
+# Classification vocabulary (#970 AC1-AC4).
+CLASS_CONFIRMED_STALE_DELETED = "confirmed_stale_deleted_worktree"
+CLASS_MOVED_WORKTREE = "moved_worktree_path"
+CLASS_UNAVAILABLE_HOST_MOUNT = "unavailable_host_or_mount"
+CLASS_LIVE_LEASE_PROTECTED = "live_lease_protected"
+CLASS_LIVE_SESSION_PROTECTED = "live_session_protected"
+CLASS_PRESENT_VALID = "present_valid_worktree"
+CLASS_ALREADY_RETIRED = "already_retired_binding"
+
+ALL_CLASSIFICATIONS = frozenset({
+    CLASS_CONFIRMED_STALE_DELETED,
+    CLASS_MOVED_WORKTREE,
+    CLASS_UNAVAILABLE_HOST_MOUNT,
+    CLASS_LIVE_LEASE_PROTECTED,
+    CLASS_LIVE_SESSION_PROTECTED,
+    CLASS_PRESENT_VALID,
+    CLASS_ALREADY_RETIRED,
+})
+
+RETIRE_ELIGIBLE_CLASSES = frozenset({
+    CLASS_CONFIRMED_STALE_DELETED,
+})
+
+
+def _norm_path(p: str | None) -> str:
+    if not p or not str(p).strip():
+        return ""
+    try:
+        return os.path.realpath(os.path.abspath(str(p).strip()))
+    except Exception:
+        return str(p).strip()
+
+
+def check_host_mount_health(project_root: str) -> dict[str, Any]:
+    """Check if the repository root and branches/ mount are healthy and accessible."""
+    root = _norm_path(project_root)
+    if not root or not os.path.isdir(root):
+        return {
+            "healthy": False,
+            "reason": f"project_root '{project_root}' does not exist or is not a directory",
+        }
+
+    # Verify project_root is inside a valid git repository
+    try:
+        res = subprocess.run(
+            ["git", "-C", root, "rev-parse", "--git-dir"],
+            capture_output=True,
+            text=True,
+            check=False,
+        )
+        if res.returncode != 0:
+            return {
+                "healthy": False,
+                "reason": f"project_root '{root}' is not a valid git repository: {res.stderr.strip()}",
+            }
+    except Exception as exc:
+        return {
+            "healthy": False,
+            "reason": f"git execution failed at project_root '{root}': {exc}",
+        }
+
+    branches_dir = os.path.join(root, "branches")
+    if os.path.exists(branches_dir) and not os.path.isdir(branches_dir):
+        return {
+            "healthy": False,
+            "reason": f"branches path '{branches_dir}' exists but is not a directory",
+        }
+
+    return {"healthy": True, "project_root": root, "branches_dir": branches_dir}
+
+
+def find_moved_worktree_path(
+    project_root: str,
+    recorded_path: str,
+    branch: str | None = None,
+) -> str | None:
+    """Check if a missing recorded worktree path is actually present at another location.
+
+    Returns the new path if found, or None if truly absent.
+    """
+    norm_rec = _norm_path(recorded_path)
+    if not norm_rec:
+        return None
+
+    wt_list = []
+    try:
+        wt_list = list_local_worktrees(project_root)
+    except Exception:
+        wt_list = []
+
+    for entry in wt_list:
+        p = _norm_path(entry.get("path"))
+        b = entry.get("branch")
+        if not p:
+            continue
+        # If git worktree list shows a worktree matching the branch at a different path
+        if branch and b and b.strip() == branch.strip() and p != norm_rec:
+            if os.path.exists(p):
+                return p
+        # If the basename matches and path exists
+        if os.path.basename(p) == os.path.basename(norm_rec) and p != norm_rec:
+            if os.path.exists(p):
+                return p
+
+    # Check potential filesystem candidates under branches/
+    branches_dir = os.path.join(_norm_path(project_root), "branches")
+    if os.path.isdir(branches_dir):
+        candidates_to_check: list[str] = []
+        if branch:
+            candidates_to_check.append(os.path.join(branches_dir, branch.replace("/", "-")))
+            candidates_to_check.append(os.path.join(branches_dir, branch))
+        if recorded_path:
+            candidates_to_check.append(os.path.join(branches_dir, os.path.basename(recorded_path)))
+
+        for cand in candidates_to_check:
+            norm_cand = _norm_path(cand)
+            if norm_cand and norm_cand != norm_rec and os.path.isdir(norm_cand):
+                return norm_cand
+
+    return None
+
+
+def classify_worktree_binding(
+    *,
+    recorded_path: str,
+    lease_status: str | None = None,
+    lease_phase: str | None = None,
+    owner_pid: int | None = None,
+    session_id: str | None = None,
+    session_active: bool = False,
+    branch: str | None = None,
+    project_root: str,
+    host_health: dict[str, Any] | None = None,
+    path_exists_override: bool | None = None,
+) -> dict[str, Any]:
+    """Classify a single worktree binding record for missing-path safety.
+
+    Fail-closed: returns detailed reasons and classification.
+    """
+    norm_p = _norm_path(recorded_path)
+    reasons: list[str] = []
+
+    if not norm_p:
+        return {
+            "classification": CLASS_ALREADY_RETIRED,
+            "retire_eligible": False,
+            "recorded_path": "",
+            "reasons": ["worktree_path is empty (already retired/unbound)"],
+        }
+
+    # Check 1: Host / mount health
+    hh = host_health or check_host_mount_health(project_root)
+    if not hh.get("healthy"):
+        reasons.append(
+            f"host or filesystem mount unavailable: {hh.get('reason')}; "
+            "refusing to classify missing path as deleted (fail closed)"
+        )
+        return {
+            "classification": CLASS_UNAVAILABLE_HOST_MOUNT,
+            "retire_eligible": False,
+            "recorded_path": norm_p,
+            "reasons": reasons,
+            "host_health": hh,
+        }
+
+    # Check 2: Path existence
+    exists = (
+        path_exists_override
+        if path_exists_override is not None
+        else os.path.exists(norm_p)
+    )
+
+    if exists:
+        return {
+            "classification": CLASS_PRESENT_VALID,
+            "retire_eligible": False,
+            "recorded_path": norm_p,
+            "reasons": [f"worktree path '{norm_p}' exists on disk"],
+        }
+
+    # Path is missing on disk. Run safety checks before declaring confirmed stale.
+
+    # Check 3: Moved path
+    moved_path = find_moved_worktree_path(project_root, norm_p, branch=branch)
+    if moved_path:
+        reasons.append(
+            f"recorded path '{norm_p}' is missing, but worktree for branch '{branch}' "
+            f"was found moved to '{moved_path}'"
+        )
+        return {
+            "classification": CLASS_MOVED_WORKTREE,
+            "retire_eligible": False,
+            "recorded_path": norm_p,
+            "moved_to_path": moved_path,
+            "reasons": reasons,
+        }
+
+    # Check 4: Live lease protection
+    st = (lease_status or "").strip().lower()
+    pid_alive = lease_lifecycle.is_process_alive(owner_pid) if owner_pid else False
+    if st == "active" and (pid_alive or owner_pid is None):
+        reasons.append(
+            f"lease is status='active' (owner_pid={owner_pid}, alive={pid_alive}); "
+            "active lease protects worktree binding"
+        )
+        return {
+            "classification": CLASS_LIVE_LEASE_PROTECTED,
+            "retire_eligible": False,
+            "recorded_path": norm_p,
+            "reasons": reasons,
+        }
+
+    # Check 5: Live session / process protection
+    if session_active and pid_alive:
+        reasons.append(
+            f"session '{session_id}' is active with live process PID {owner_pid}; "
+            "live session protects worktree binding"
+        )
+        return {
+            "classification": CLASS_LIVE_SESSION_PROTECTED,
+            "retire_eligible": False,
+            "recorded_path": norm_p,
+            "reasons": reasons,
+        }
+
+    # Confirmed stale deleted
+    reasons.append(
+        f"worktree path '{norm_p}' does not exist on disk, host is healthy, "
+        "no moved path was found, and no active lease or live process references it"
+    )
+    return {
+        "classification": CLASS_CONFIRMED_STALE_DELETED,
+        "retire_eligible": True,
+        "recorded_path": norm_p,
+        "reasons": reasons,
+    }
+
+
+def audit_missing_worktree_bindings(
+    db: cpd.ControlPlaneDB | None = None,
+    *,
+    project_root: str | None = None,
+    remote: str = "prgs",
+    org: str | None = None,
+    repo: str | None = None,
+    host: str | None = None,
+) -> dict[str, Any]:
+    """Audit all recorded worktree bindings and classify missing-path entries.
+
+    Read-only / side-effect free.
+    """
+    root = _norm_path(project_root or os.getcwd())
+    hh = check_host_mount_health(root)
+    cp_db = db or cpd.ControlPlaneDB()
+
+    bindings: list[dict[str, Any]] = []
+
+    # 1. Audit control-plane leases
+    try:
+        leases_res = lease_lifecycle.list_active_leases(
+            cp_db,
+            remote=remote,
+            org=org,
+            repo=repo,
+            include_non_active=True,
+            limit=500,
+        )
+        for L in leases_res.get("leases") or []:
+            wt = (L.get("worktree_path") or "").strip()
+            if not wt:
+                continue
+            owner_pid = L.get("owner_pid") or L.get("session_pid")
+            sess_id = L.get("session_id")
+            br = L.get("branch")
+            if not br and isinstance(L.get("provenance"), dict):
+                br = L["provenance"].get("branch")
+
+            cls_info = classify_worktree_binding(
+                recorded_path=wt,
+                lease_status=L.get("status"),
+                lease_phase=L.get("phase"),
+                owner_pid=owner_pid,
+                session_id=sess_id,
+                branch=br,
+                project_root=root,
+                host_health=hh,
+            )
+
+            bindings.append({
+                "source": "lease",
+                "lease_id": L.get("lease_id"),
+                "session_id": sess_id,
+                "work_kind": L.get("work_kind"),
+                "work_number": L.get("work_number"),
+                "branch": br,
+                "worktree_path": wt,
+                "lease_status": L.get("status"),
+                "lease_phase": L.get("phase"),
+                "owner_pid": owner_pid,
+                "remote": L.get("remote") or remote,
+                "org": L.get("org") or org,
+                "repo": L.get("repo") or repo,
+                "host": L.get("host") or host,
+                **cls_info,
+            })
+    except Exception as exc:  # noqa: BLE001
+        bindings.append({
+            "source": "lease_query_error",
+            "error": str(exc),
+        })
+
+    # 2. Audit session checkpoints
+    try:
+        checkpoints = cp_db.list_session_checkpoints(limit=500)
+        for cp in checkpoints:
+            wt = (cp.get("worktree_path") or "").strip()
+            if not wt:
+                continue
+            sess_id = cp.get("session_id")
+            cls_info = classify_worktree_binding(
+                recorded_path=wt,
+                session_id=sess_id,
+                branch=cp.get("branch"),
+                project_root=root,
+                host_health=hh,
+            )
+            bindings.append({
+                "source": "session_checkpoint",
+                "checkpoint_id": cp.get("checkpoint_id"),
+                "session_id": sess_id,
+                "work_kind": cp.get("work_kind"),
+                "work_number": cp.get("work_number"),
+                "branch": cp.get("branch"),
+                "worktree_path": wt,
+                "status": cp.get("status"),
+                "remote": cp.get("remote") or remote,
+                "org": cp.get("org") or org,
+                "repo": cp.get("repo") or repo,
+                **cls_info,
+            })
+    except Exception:  # noqa: BLE001
+        pass
+
+    missing_bindings = [
+        b for b in bindings if b.get("classification") != CLASS_PRESENT_VALID
+    ]
+    confirmed_stale = [
+        b for b in missing_bindings if b.get("classification") == CLASS_CONFIRMED_STALE_DELETED
+    ]
+    moved = [
+        b for b in missing_bindings if b.get("classification") == CLASS_MOVED_WORKTREE
+    ]
+    live_protected = [
+        b for b in missing_bindings if b.get("classification") in (CLASS_LIVE_LEASE_PROTECTED, CLASS_LIVE_SESSION_PROTECTED)
+    ]
+    unavailable_host = [
+        b for b in missing_bindings if b.get("classification") == CLASS_UNAVAILABLE_HOST_MOUNT
+    ]
+
+    return {
+        "project_root": root,
+        "host_mount_healthy": hh.get("healthy", False),
+        "host_health_reason": hh.get("reason"),
+        "total_bindings_audited": len(bindings),
+        "present_count": len(bindings) - len(missing_bindings),
+        "missing_count": len(missing_bindings),
+        "confirmed_stale_count": len(confirmed_stale),
+        "moved_count": len(moved),
+        "live_protected_count": len(live_protected),
+        "unavailable_host_count": len(unavailable_host),
+        "missing_bindings": missing_bindings,
+        "confirmed_stale_candidates": confirmed_stale,
+        "worktrees_dimension_resolved": len(missing_bindings) == 0,
+    }
+
+
+def resolve_missing_worktree_binding(
+    db: cpd.ControlPlaneDB | None = None,
+    *,
+    binding: dict[str, Any],
+    dry_run: bool = True,
+    operator_authorized: bool = False,
+    workflow_authorized: bool = False,
+    project_root: str | None = None,
+) -> dict[str, Any]:
+    """Safely resolve (retire) one confirmed stale missing worktree binding (#970).
+
+    Pre-mutation re-validation runs immediately before mutation to prevent races.
+    """
+    root = _norm_path(project_root or os.getcwd())
+    cp_db = db or cpd.ControlPlaneDB()
+
+    rec_path = binding.get("worktree_path") or binding.get("recorded_path") or ""
+    if not rec_path:
+        return {
+            "success": True,
+            "performed": False,
+            "outcome": "already_retired",
+            "message": "worktree_path is already empty (no action needed)",
+        }
+
+    # 1. Authorization check
+    authorized = operator_authorized or workflow_authorized
+    if not authorized and not dry_run:
+        return {
+            "success": False,
+            "performed": False,
+            "reason": "authorization_required",
+            "message": "Mutation refused: operator_authorized or workflow_authorized is required to retire a stale binding",
+        }
+
+    # 2. Immediate pre-mutation re-validation
+    reval = classify_worktree_binding(
+        recorded_path=rec_path,
+        lease_status=binding.get("lease_status"),
+        lease_phase=binding.get("lease_phase"),
+        owner_pid=binding.get("owner_pid"),
+        session_id=binding.get("session_id"),
+        branch=binding.get("branch"),
+        project_root=root,
+    )
+
+    if reval.get("classification") != CLASS_CONFIRMED_STALE_DELETED:
+        return {
+            "success": False,
+            "performed": False,
+            "reason": "revalidation_failed",
+            "revalidation": reval,
+            "message": f"Pre-mutation re-validation failed: binding classified as '{reval.get('classification')}' instead of '{CLASS_CONFIRMED_STALE_DELETED}'",
+        }
+
+    source = binding.get("source", "lease")
+    lease_id = binding.get("lease_id")
+    session_id = binding.get("session_id")
+    checkpoint_id = binding.get("checkpoint_id")
+
+    planned_action = {
+        "source": source,
+        "lease_id": lease_id,
+        "session_id": session_id,
+        "checkpoint_id": checkpoint_id,
+        "retired_worktree_path": rec_path,
+        "classification": CLASS_CONFIRMED_STALE_DELETED,
+    }
+
+    if dry_run:
+        return {
+            "success": True,
+            "performed": False,
+            "dry_run": True,
+            "action": "retire_stale_binding",
+            "planned_action": planned_action,
+            "message": f"Dry-run: would retire stale missing worktree_path '{rec_path}' for {source} (lease={lease_id})",
+        }
+
+    # 3. Apply mode: execute targeted retirement
+    audit_proof: dict[str, Any] = {
+        "retired_worktree_path": rec_path,
+        "source": source,
+        "lease_id": lease_id,
+        "session_id": session_id,
+        "checkpoint_id": checkpoint_id,
+        "classification": CLASS_CONFIRMED_STALE_DELETED,
+    }
+
+    if source == "lease" and lease_id:
+        ret_proof = cp_db.retire_lease_worktree_path(
+            lease_id,
+            expected_path=rec_path,
+            reason="missing_worktree_path_retired_by_reconciler",
+        )
+        audit_proof["db_lease_update"] = ret_proof
+
+    if source == "session_checkpoint" and (checkpoint_id or session_id):
+        ret_cp = cp_db.retire_session_checkpoint_worktree_path(
+            session_id=session_id or "",
+            checkpoint_id=checkpoint_id,
+            expected_path=rec_path,
+            reason="missing_worktree_path_retired_by_reconciler",
+        )
+        audit_proof["db_checkpoint_update"] = ret_cp
+
+    # Also clean up any matching lease row if checkpoint was supplied or vice versa
+    if lease_id and source != "lease":
+        try:
+            cp_db.retire_lease_worktree_path(lease_id, expected_path=rec_path)
+        except Exception:  # noqa: BLE001
+            pass
+
+    return {
+        "success": True,
+        "performed": True,
+        "dry_run": False,
+        "action": "retired_stale_binding",
+        "audit_proof": audit_proof,
+        "message": f"Successfully retired stale missing worktree_path '{rec_path}' for lease {lease_id}",
+    }
+
+
+def reconcile_missing_worktree_bindings(
+    db: cpd.ControlPlaneDB | None = None,
+    *,
+    project_root: str | None = None,
+    remote: str = "prgs",
+    org: str | None = None,
+    repo: str | None = None,
+    host: str | None = None,
+    dry_run: bool = True,
+    operator_authorized: bool = False,
+    workflow_authorized: bool = False,
+) -> dict[str, Any]:
+    """Audit and safely reconcile missing worktree path bindings (#970).
+
+    Audits all recorded bindings, identifies confirmed stale missing paths,
+    retires eligible bindings (when dry_run=False and authorized), and returns
+    durable before/after audit evidence.
+    """
+    cp_db = db or cpd.ControlPlaneDB()
+    root = _norm_path(project_root or os.getcwd())
+
+    initial_audit = audit_missing_worktree_bindings(
+        cp_db,
+        project_root=root,
+        remote=remote,
+        org=org,
+        repo=repo,
+        host=host,
+    )
+
+    candidates = initial_audit.get("confirmed_stale_candidates") or []
+    resolutions: list[dict[str, Any]] = []
+
+    for candidate in candidates:
+        res = resolve_missing_worktree_binding(
+            cp_db,
+            binding=candidate,
+            dry_run=dry_run,
+            operator_authorized=operator_authorized,
+            workflow_authorized=workflow_authorized,
+            project_root=root,
+        )
+        resolutions.append(res)
+
+    post_audit = audit_missing_worktree_bindings(
+        cp_db,
+        project_root=root,
+        remote=remote,
+        org=org,
+        repo=repo,
+        host=host,
+    )
+
+    return {
+        "success": True,
+        "dry_run": dry_run,
+        "operator_authorized": operator_authorized,
+        "workflow_authorized": workflow_authorized,
+        "before_audit": initial_audit,
+        "resolutions": resolutions,
+        "after_audit": post_audit,
+        "worktrees_dimension_resolved": post_audit.get("worktrees_dimension_resolved", False),
+        "stale_candidates_count": len(candidates),
+        "resolved_count": sum(1 for r in resolutions if r.get("performed") or (dry_run and r.get("success"))),
+    }

tests/test_issue_970_missing_worktree_reconcile.py

@@ -0,0 +1,243 @@
+"""Tests for issue #970: safely resolve worktree bindings whose paths are missing.
+
+Covers all acceptance criteria for #970:
+- missing path reporting with repo/branch/issue/session/lease/host correlation
+- fail closed on unavailable host or mount failure
+- fail closed on moved worktree paths
+- fail closed on live lease or active session/process
+- atomic pre-mutation re-validation preventing recreation races
+- targeted retirement preserving unrelated worktrees & metadata
+- safe and idempotent repeated execution
+- worktrees dimension resolution post-cleanup
+- resolution of the two historical post-PR #968 records (#795 / #635)
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import shutil
+import tempfile
+import unittest
+
+import control_plane_db as cpd
+import lease_lifecycle
+import missing_worktree_reconcile as mwr
+
+
+class TestMissingWorktreeReconcile(unittest.TestCase):
+    def setUp(self):
+        self.temp_dir = tempfile.mkdtemp()
+        self.db_path = os.path.join(self.temp_dir, "control_plane.sqlite3")
+        self.db = cpd.ControlPlaneDB(self.db_path)
+        self.project_root = os.path.join(self.temp_dir, "repo")
+        os.makedirs(os.path.join(self.project_root, "branches"), exist_ok=True)
+        # Initialize git repo in project_root for git rev-parse checks
+        os.system(f"git -C '{self.project_root}' init -q")
+
+    def tearDown(self):
+        shutil.rmtree(self.temp_dir, ignore_errors=True)
+
+    def test_audit_missing_worktree_bindings_reports_associations(self):
+        res = self.db.assign_and_lease(
+            remote="prgs",
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+            kind="pr",
+            number=795,
+            expected_head_sha="head795",
+            role="reviewer",
+            session_id="reviewer-pr795-test-session",
+            worktree_path="/tmp/nonexistent_wt_path_795",
+        )
+        lease_id = res.lease_id
+        self.assertIsNotNone(lease_id)
+        # Release the lease (dead/released)
+        self.db.release_lease_recorded(lease_id, session_id="reviewer-pr795-test-session")
+
+        audit = mwr.audit_missing_worktree_bindings(
+            self.db,
+            project_root=self.project_root,
+            remote="prgs",
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+        )
+
+        self.assertTrue(audit["host_mount_healthy"])
+        self.assertEqual(audit["missing_count"], 1)
+        self.assertEqual(audit["confirmed_stale_count"], 1)
+
+        candidate = audit["confirmed_stale_candidates"][0]
+        self.assertEqual(candidate["lease_id"], lease_id)
+        self.assertEqual(candidate["work_number"], 795)
+        self.assertEqual(candidate["session_id"], "reviewer-pr795-test-session")
+        self.assertEqual(candidate["classification"], mwr.CLASS_CONFIRMED_STALE_DELETED)
+        self.assertTrue(candidate["retire_eligible"])
+
+    def test_unavailable_host_mount_blocks_retirement(self):
+        # Point to a non-existent project_root
+        bad_root = os.path.join(self.temp_dir, "nonexistent_repo_dir")
+        audit = mwr.audit_missing_worktree_bindings(
+            self.db,
+            project_root=bad_root,
+            remote="prgs",
+        )
+        self.assertFalse(audit["host_mount_healthy"])
+
+        res = self.db.assign_and_lease(
+            remote="prgs", org="org", repo="repo", kind="issue", number=100,
+            role="author", session_id="sess-100", worktree_path="/tmp/missing_100",
+        )
+        self.db.release_lease_recorded(res.lease_id, session_id="sess-100")
+
+        audit2 = mwr.audit_missing_worktree_bindings(self.db, project_root=bad_root, remote="prgs")
+        self.assertEqual(audit2["unavailable_host_count"], 1)
+        self.assertEqual(audit2["confirmed_stale_count"], 0)
+        self.assertFalse(audit2["missing_bindings"][0]["retire_eligible"])
+
+    def test_live_lease_and_live_session_block_retirement(self):
+        res = self.db.assign_and_lease(
+            remote="prgs", org="org", repo="repo", kind="issue", number=200,
+            role="author", session_id="sess-live-200", worktree_path="/tmp/missing_live_200",
+            owner_pid=os.getpid(),
+        )
+
+        audit = mwr.audit_missing_worktree_bindings(self.db, project_root=self.project_root, remote="prgs")
+        self.assertEqual(audit["live_protected_count"], 1)
+        self.assertEqual(audit["confirmed_stale_count"], 0)
+
+        # Attempting resolution on live protected binding is refused
+        c = audit["missing_bindings"][0]
+        res_mut = mwr.resolve_missing_worktree_binding(self.db, binding=c, dry_run=False, operator_authorized=True, project_root=self.project_root)
+        self.assertFalse(res_mut["success"])
+        self.assertEqual(res_mut["reason"], "revalidation_failed")
+
+    def test_moved_worktree_path_blocks_retirement(self):
+        # Create a worktree under branches/ to simulate moved path
+        moved_dir = os.path.join(self.project_root, "branches", "issue-635-project-registry-api")
+        os.makedirs(moved_dir, exist_ok=True)
+
+        res = self.db.assign_and_lease(
+            remote="prgs", org="org", repo="repo", kind="issue", number=635,
+            role="author", session_id="sess-635", worktree_path="/tmp/old_missing_path_635",
+        )
+        self.db.release_lease_recorded(res.lease_id, session_id="sess-635")
+
+        cls_info = mwr.classify_worktree_binding(
+            recorded_path="/tmp/old_missing_path_635",
+            branch="issue-635-project-registry-api",
+            project_root=self.project_root,
+        )
+        self.assertEqual(cls_info["classification"], mwr.CLASS_MOVED_WORKTREE)
+        self.assertFalse(cls_info["retire_eligible"])
+        self.assertEqual(os.path.realpath(cls_info["moved_to_path"]), os.path.realpath(moved_dir))
+
+    def test_recreation_race_revalidation_blocks_mutation(self):
+        path = os.path.join(self.temp_dir, "recreated_wt")
+        res = self.db.assign_and_lease(
+            remote="prgs", org="org", repo="repo", kind="issue", number=300,
+            role="author", session_id="sess-300", worktree_path=path,
+        )
+        self.db.release_lease_recorded(res.lease_id, session_id="sess-300")
+
+        # Initial audit when path is missing
+        audit = mwr.audit_missing_worktree_bindings(self.db, project_root=self.project_root, remote="prgs")
+        candidate = audit["confirmed_stale_candidates"][0]
+
+        # Simulate concurrent recreation of worktree path
+        os.makedirs(path, exist_ok=True)
+
+        # Pre-mutation revalidation detects recreation and blocks
+        res_mut = mwr.resolve_missing_worktree_binding(
+            self.db,
+            binding=candidate,
+            dry_run=False,
+            operator_authorized=True,
+            project_root=self.project_root,
+        )
+        self.assertFalse(res_mut["success"])
+        self.assertEqual(res_mut["reason"], "revalidation_failed")
+
+    def test_authorized_cleanup_retires_stale_binding_and_is_idempotent(self):
+        # Create 2 leases: 1 stale missing, 1 valid existing worktree
+        valid_path = os.path.join(self.project_root, "branches", "valid_wt")
+        os.makedirs(valid_path, exist_ok=True)
+
+        stale_path = "/tmp/stale_missing_400"
+
+        res1 = self.db.assign_and_lease(
+            remote="prgs", org="org", repo="repo", kind="issue", number=401,
+            role="author", session_id="sess-401", worktree_path=valid_path,
+        )
+
+        res2 = self.db.assign_and_lease(
+            remote="prgs", org="org", repo="repo", kind="issue", number=402,
+            role="author", session_id="sess-402", worktree_path=stale_path,
+        )
+        self.db.release_lease_recorded(res2.lease_id, session_id="sess-402")
+
+        # Dry-run reconciliation
+        dry = mwr.reconcile_missing_worktree_bindings(
+            self.db, project_root=self.project_root, remote="prgs", dry_run=True, operator_authorized=True
+        )
+        self.assertTrue(dry["success"])
+        self.assertTrue(dry["dry_run"])
+        self.assertEqual(dry["stale_candidates_count"], 1)
+
+        # Apply reconciliation
+        applied = mwr.reconcile_missing_worktree_bindings(
+            self.db, project_root=self.project_root, remote="prgs", dry_run=False, operator_authorized=True
+        )
+        self.assertTrue(applied["success"])
+        self.assertFalse(applied["dry_run"])
+        self.assertTrue(applied["worktrees_dimension_resolved"])
+        self.assertEqual(applied["after_audit"]["missing_count"], 0)
+
+        # Verify DB: lease 2 worktree_path is cleared, lease 1 worktree_path remains intact
+        l1 = self.db.get_lease_workflow_state(res1.lease_id)
+        self.assertEqual(l1["lease"]["worktree_path"], valid_path)
+
+        l2 = self.db.get_lease_workflow_state(res2.lease_id)
+        self.assertEqual(l2["lease"]["worktree_path"], "")
+        prov = json.loads(l2["lease"]["provenance_json"])
+        self.assertTrue(prov.get("worktree_path_retired"))
+        self.assertEqual(prov.get("retired_worktree_path"), stale_path)
+
+        # Re-run (idempotency check)
+        rerun = mwr.reconcile_missing_worktree_bindings(
+            self.db, project_root=self.project_root, remote="prgs", dry_run=False, operator_authorized=True
+        )
+        self.assertTrue(rerun["success"])
+        self.assertTrue(rerun["worktrees_dimension_resolved"])
+        self.assertEqual(rerun["stale_candidates_count"], 0)
+
+    def test_historical_post_pr968_records_resolution(self):
+        # Create historical records matching #970 description
+        res_795 = self.db.assign_and_lease(
+            remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", kind="pr", number=795,
+            expected_head_sha="head795hist", role="reviewer", session_id="reviewer-pr795-hist",
+            worktree_path="/Users/jasonwalker/Development/Gitea-Tools/branches/review-feat-issue-628-autonomous-handoffs-orchestration",
+        )
+        self.db.release_lease_recorded(res_795.lease_id, session_id="reviewer-pr795-hist")
+
+        res_635 = self.db.assign_and_lease(
+            remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", kind="issue", number=635,
+            role="author", session_id="author-issue635-hist",
+            worktree_path="/Users/jasonwalker/Development/Gitea-Tools/branches/issue-635-project-registry-api",
+        )
+        self.db.release_lease_recorded(res_635.lease_id, session_id="author-issue635-hist")
+
+        # Audit finds both historical records
+        audit = mwr.audit_missing_worktree_bindings(self.db, project_root=self.project_root, remote="prgs")
+        self.assertEqual(audit["confirmed_stale_count"], 2)
+
+        # Reconcile safely resolves both
+        rec = mwr.reconcile_missing_worktree_bindings(
+            self.db, project_root=self.project_root, remote="prgs", dry_run=False, operator_authorized=True
+        )
+        self.assertTrue(rec["worktrees_dimension_resolved"])
+        self.assertEqual(rec["after_audit"]["missing_count"], 0)
+
+
+if __name__ == "__main__":
+    unittest.main()