control_plane_db.py

@@ -280,6 +280,22 @@ def _ts(dt: datetime | None = None) -> str:
     return value.astimezone(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
 
 
+def _realpath_or_raw(value: str | None) -> str:
+    """Normalize a filesystem path for compare-and-swap equality (#970).
+
+    Symlinks and ``..`` segments must not make two spellings of the same path
+    look different, but an unresolvable path must still compare as itself
+    rather than collapsing to empty — an empty result means "no path given".
+    """
+    text = (value or "").strip()
+    if not text:
+        return ""
+    try:
+        return os.path.realpath(os.path.abspath(text))
+    except Exception:
+        return text
+
+
 def _parse_ts(value: str | None) -> datetime | None:
     if not value:
         return None
@@ -1913,6 +1929,168 @@ class ControlPlaneDB:
                 ),
             )
 
+    def retire_lease_worktree_path(
+        self,
+        lease_id: str,
+        *,
+        expected_path: str | None = None,
+        expected_status: str | None = None,
+        expected_session_id: str | None = None,
+        expected_owner_pid: int | 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.
+
+        The update is a compare-and-swap (#970 review 644 B1/B3): the caller
+        states the exact path it audited and, when known, the lease status,
+        owning session, and owner pid it classified against. Every stated value
+        must still match the stored row, and the ``UPDATE`` itself is keyed on
+        the stored ``worktree_path``, so a concurrent writer that moved or
+        replaced the binding between audit and apply loses the race instead of
+        having its value silently overwritten. A mismatch raises and mutates
+        nothing.
+
+        ``expected_path`` is mandatory: a retirement that does not name the path
+        it intends to clear cannot be safe against concurrent recreation.
+        """
+        now_s = _ts()
+        expected_norm = _realpath_or_raw(expected_path)
+        if not expected_norm:
+            raise ControlPlaneError(
+                f"cannot retire lease {lease_id} worktree_path: expected_path is "
+                "required for compare-and-swap retirement (fail closed)"
+            )
+        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}")
+
+            record = dict(row)
+            current_wt = (record.get("worktree_path") or "").strip()
+
+            if not current_wt:
+                # Idempotent: the binding this caller audited is already gone.
+                return {
+                    "lease_id": lease_id,
+                    "retired": False,
+                    "already_retired": True,
+                    "prior_worktree_path": "",
+                    "expected_worktree_path": expected_path,
+                    "reason": reason,
+                    "compare_and_swap": {
+                        "matched": True,
+                        "outcome": "already_retired",
+                    },
+                }
+
+            if _realpath_or_raw(current_wt) != expected_norm:
+                raise ControlPlaneError(
+                    f"cannot retire lease {lease_id} worktree_path: expected "
+                    f"'{expected_path}' does not match current '{current_wt}' "
+                    "(fail closed)"
+                )
+
+            for field, expected_value in (
+                ("status", expected_status),
+                ("session_id", expected_session_id),
+            ):
+                if expected_value is None:
+                    continue
+                current_value = record.get(field)
+                if str(current_value or "").strip() != str(expected_value).strip():
+                    raise ControlPlaneError(
+                        f"cannot retire lease {lease_id} worktree_path: lease "
+                        f"{field} changed since audit (expected "
+                        f"'{expected_value}', found '{current_value}'); fail closed"
+                    )
+
+            if expected_owner_pid is not None:
+                current_pid = record.get("owner_pid")
+                if current_pid is not None and int(current_pid) != int(expected_owner_pid):
+                    raise ControlPlaneError(
+                        f"cannot retire lease {lease_id} worktree_path: lease "
+                        f"owner_pid changed since audit (expected "
+                        f"{expected_owner_pid}, found {current_pid}); fail closed"
+                    )
+
+            # Parse and update provenance_json
+            raw_prov = record.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
+            prov.update({
+                "worktree_path_retired": True,
+                "retired_worktree_path": prior_path,
+                "retired_at": now_s,
+                "retirement_reason": reason,
+                "retired_from_status": record.get("status"),
+                "retired_from_session_id": record.get("session_id"),
+                "worktree_path": "",
+            })
+            prov_json = json.dumps(prov)
+
+            if "worktree_path" in cols:
+                # CAS: keyed on the exact stored path this caller audited.
+                cur = conn.execute(
+                    "UPDATE leases SET worktree_path = '', provenance_json = ? "
+                    "WHERE lease_id = ? AND worktree_path = ?",
+                    (prov_json, lease_id, record.get("worktree_path")),
+                )
+                if cur.rowcount != 1:
+                    raise ControlPlaneError(
+                        f"cannot retire lease {lease_id} worktree_path: "
+                        "compare-and-swap matched no row (concurrent change); "
+                        "fail closed"
+                    )
+            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', ?, ?)
+                """,
+                (
+                    record["work_item_id"],
+                    f"lease {lease_id} worktree_path '{prior_path}' retired: {reason}",
+                    now_s,
+                ),
+            )
+
+            return {
+                "lease_id": lease_id,
+                "retired": True,
+                "already_retired": False,
+                "prior_worktree_path": prior_path,
+                "expected_worktree_path": expected_path,
+                "retired_at": now_s,
+                "reason": reason,
+                "compare_and_swap": {
+                    "matched": True,
+                    "outcome": "retired",
+                    "expected_status": expected_status,
+                    "expected_session_id": expected_session_id,
+                    "expected_owner_pid": expected_owner_pid,
+                },
+            }
+
+
     def abandon_lease(
         self,
         *,
@@ -3051,3 +3229,123 @@ 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,
+        expected_status: str | None = None,
+        reason: str = "missing_worktree_path_retired",
+    ) -> dict[str, Any]:
+        """Retire a missing worktree_path from session_checkpoints (#970).
+
+        Compare-and-swap, mirroring :meth:`retire_lease_worktree_path` (#970
+        review 644 B3). ``expected_path`` names the exact stored path the caller
+        audited; the guarded ``UPDATE`` is keyed on that stored value, so a
+        checkpoint whose path was moved, replaced, or concurrently rewritten
+        after the audit is refused without mutation rather than blindly cleared.
+
+        Exactly one checkpoint row is targeted: by ``checkpoint_id`` when given,
+        otherwise by ``session_id``, which must identify a single row.
+        """
+        now_s = _ts()
+        expected_norm = _realpath_or_raw(expected_path)
+        if not expected_norm:
+            raise ControlPlaneError(
+                "cannot retire session checkpoint worktree_path: expected_path "
+                "is required for compare-and-swap retirement (fail closed)"
+            )
+        if not checkpoint_id and not (session_id or "").strip():
+            raise ControlPlaneError(
+                "cannot retire session checkpoint worktree_path: checkpoint_id "
+                "or session_id is required (fail closed)"
+            )
+
+        with self._tx() as conn:
+            if checkpoint_id:
+                selector_sql = "SELECT * FROM session_checkpoints WHERE checkpoint_id = ?"
+                selector_params: tuple[Any, ...] = (checkpoint_id,)
+                selector_desc = f"checkpoint_id '{checkpoint_id}'"
+            else:
+                selector_sql = "SELECT * FROM session_checkpoints WHERE session_id = ?"
+                selector_params = (session_id,)
+                selector_desc = f"session_id '{session_id}'"
+
+            rows = [dict(r) for r in conn.execute(selector_sql, selector_params).fetchall()]
+            if not rows:
+                raise ControlPlaneError(
+                    f"cannot retire session checkpoint worktree_path: no "
+                    f"checkpoint matches {selector_desc} (fail closed)"
+                )
+            if len(rows) > 1:
+                raise ControlPlaneError(
+                    f"cannot retire session checkpoint worktree_path: "
+                    f"{selector_desc} matches {len(rows)} checkpoints; supply an "
+                    "exact checkpoint_id (fail closed)"
+                )
+
+            record = rows[0]
+            target_checkpoint_id = record.get("checkpoint_id")
+            current_wt = (record.get("worktree_path") or "").strip()
+
+            if not current_wt:
+                # Idempotent: the binding this caller audited is already gone.
+                return {
+                    "session_id": session_id,
+                    "checkpoint_id": target_checkpoint_id,
+                    "retired": False,
+                    "already_retired": True,
+                    "prior_worktree_path": "",
+                    "expected_worktree_path": expected_path,
+                    "reason": reason,
+                    "compare_and_swap": {
+                        "matched": True,
+                        "outcome": "already_retired",
+                    },
+                }
+
+            if _realpath_or_raw(current_wt) != expected_norm:
+                raise ControlPlaneError(
+                    f"cannot retire session checkpoint worktree_path for "
+                    f"{selector_desc}: expected '{expected_path}' does not match "
+                    f"current '{current_wt}' (fail closed)"
+                )
+
+            if expected_status is not None:
+                current_status = record.get("status")
+                if str(current_status or "").strip() != str(expected_status).strip():
+                    raise ControlPlaneError(
+                        f"cannot retire session checkpoint worktree_path for "
+                        f"{selector_desc}: status changed since audit (expected "
+                        f"'{expected_status}', found '{current_status}'); fail closed"
+                    )
+
+            cur = conn.execute(
+                "UPDATE session_checkpoints SET worktree_path = '', updated_at = ? "
+                "WHERE checkpoint_id = ? AND worktree_path = ?",
+                (now_s, target_checkpoint_id, record.get("worktree_path")),
+            )
+            if cur.rowcount != 1:
+                raise ControlPlaneError(
+                    f"cannot retire session checkpoint worktree_path for "
+                    f"{selector_desc}: compare-and-swap matched no row "
+                    "(concurrent change); fail closed"
+                )
+
+            return {
+                "session_id": session_id,
+                "checkpoint_id": target_checkpoint_id,
+                "retired": True,
+                "already_retired": False,
+                "prior_worktree_path": current_wt,
+                "expected_worktree_path": expected_path,
+                "retired_at": now_s,
+                "reason": reason,
+                "compare_and_swap": {
+                    "matched": True,
+                    "outcome": "retired",
+                    "expected_status": expected_status,
+                },
+            }

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,162 @@ 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,
+    # Deprecated: retained so callers that still pass it get an explicit deny.
+    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
+    against a fresh authoritative read immediately before mutation to prevent
+    recreation and ownership races, and retires only the exact stale bindings
+    while preserving unrelated worktrees and Git metadata.
+
+    Apply mode (``dry_run=False``) is authorized **server-side** (#970 review
+    644 B2): it requires the reconciler-only ``gitea.branch.delete``
+    capability, the resolved ``reconcile_missing_worktree_bindings`` task
+    capability, and an active cleanup phase minted through
+    ``gitea_authorize_reconciliation_cleanup_phase``. A caller-supplied
+    ``operator_authorized`` is never authorization evidence (#709 F1 /
+    review 434) and is rejected outright. Dry-run stays available to any
+    ``gitea.read`` profile and never mutates.
+
+    Args:
+        dry_run: If True (default), reports planned mutations without modifying state.
+        operator_authorized: Rejected. Authorization is a server-side artifact.
+        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.
+    """
+    import missing_worktree_reconcile
+
+    # Explicitly reject the self-assertable Boolean before anything else, so it
+    # can never combine with a legitimate gate to authorize a mutation.
+    if operator_authorized:
+        return {
+            "success": False,
+            "performed": False,
+            "reason": "operator_authorized_rejected",
+            "reasons": [missing_worktree_reconcile.OPERATOR_AUTHORIZED_REJECTION],
+            "exact_next_action": (
+                "authorize cleanup via gitea_authorize_reconciliation_cleanup_phase "
+                "from a reconciler profile, then re-run with dry_run=False"
+            ),
+        }
+
+    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"),
+        }
+
+    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 profile.get("role_kind") or "").strip().lower()
+
+    cleanup_authorization = None
+    if not dry_run:
+        cleanup_task = missing_worktree_reconcile.CLEANUP_TASK
+        required_permission = task_capability_map.required_permission(cleanup_task)
+        capability_blockers = _profile_operation_gate(required_permission)
+        role_matches = role == task_capability_map.required_role(cleanup_task)
+        cleanup_authorization = missing_worktree_reconcile.assess_cleanup_authorization(
+            role=role,
+            capability_blockers=capability_blockers,
+            operator_authorized=False,
+            task_capability_resolved=(not capability_blockers) and role_matches,
+        )
+        if not cleanup_authorization.get("authorized"):
+            return {
+                "success": False,
+                "performed": False,
+                "reason": "cleanup_authorization_required",
+                "reasons": cleanup_authorization.get("reasons") or [],
+                "cleanup_authorization": cleanup_authorization,
+                "permission_report": _permission_block_report(required_permission),
+                "exact_next_action": (
+                    "resolve the reconcile_missing_worktree_bindings capability "
+                    "from a reconciler profile and authorize cleanup via "
+                    "gitea_authorize_reconciliation_cleanup_phase"
+                ),
+            }
+
+    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=False,
+        cleanup_authorization=cleanup_authorization,
+    )
+
+
 @mcp.tool()
 def gitea_audit_worktree_cleanup(
     remote: str = "dadeschools",

task_capability_map.py

@@ -386,6 +386,25 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
         "permission": "gitea.branch.delete",
         "role": "reconciler",
     },
+    # #970: auditing missing worktree bindings is read-only; retiring one is a
+    # control-plane cleanup mutation and carries the same reconciler-only
+    # authority as any other reconciliation cleanup (review 644 B2).
+    "audit_missing_worktree_bindings": {
+        "permission": "gitea.read",
+        "role": "reconciler",
+    },
+    "gitea_audit_missing_worktree_bindings": {
+        "permission": "gitea.read",
+        "role": "reconciler",
+    },
+    "reconcile_missing_worktree_bindings": {
+        "permission": "gitea.branch.delete",
+        "role": "reconciler",
+    },
+    "gitea_reconcile_missing_worktree_bindings": {
+        "permission": "gitea.branch.delete",
+        "role": "reconciler",
+    },
     "work_issue": {
         "permission": "gitea.pr.create",
         "role": "author",
@@ -653,6 +672,8 @@ ROLE_EXCLUSIVE_TASKS: frozenset[str] = frozenset(
         "delete_branch",
         "cleanup_merged_pr_branch",
         "reconciliation_cleanup",
+        "reconcile_missing_worktree_bindings",
+        "gitea_reconcile_missing_worktree_bindings",
         "work_issue",
         "work-issue",
     }

tests/test_task_capability_role_invariants.py

@@ -153,6 +153,12 @@ EXPECTED_ROLE_EXCLUSIVE_TASKS = frozenset(
         "delete_branch",
         "cleanup_merged_pr_branch",
         "reconciliation_cleanup",
+        # #970 review 644 B2: retiring a missing worktree binding is a
+        # control-plane cleanup mutation, so it carries the same reconciler-only
+        # authority as every other reconciliation cleanup. Permission alone must
+        # not authorize it.
+        "reconcile_missing_worktree_bindings",
+        "gitea_reconcile_missing_worktree_bindings",
         "work_issue",
         "work-issue",
     }