Merge branch 'master' into fix/issue-850-native-mcp-bootstrap

Post-merge cleanup previously continued past ownership-blocked remote
deletes, which skipped independently safe local worktree removal when the
only block was worktree_binding. Remove the clean owned worktree first,
reassess ownership, then delete the remote branch only if still safe.
Preserve fail-closed protection for dirty/foreign ownership categories.

Closes #851

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>

allocator_service.py

@@ -1216,6 +1216,12 @@ def allocate_next_work(
             "reasons": [
                 "dry-run only (apply=false); no assignment/lease created — "
                 "call again with apply=true to reserve via control-plane DB"
+                + (
+                    "; after apply, the required-role worker consumes via "
+                    "gitea_adopt_workflow_lease (#843)"
+                    if mode == ALLOCATION_MODE_CROSS_ROLE and expected_role != role_norm
+                    else ""
+                )
             ],
             "skipped": [s.as_dict() for s in skipped],
             "terminal_pr": terminal_pr,
@@ -1247,6 +1253,9 @@ def allocate_next_work(
     # Atomic reserve via #613 substrate.
     ttl = lease_ttl_seconds if lease_ttl_seconds is not None else None
     try:
+        cross_role_handoff = (
+            mode == ALLOCATION_MODE_CROSS_ROLE and lease_role != role_norm
+        )
         kwargs: dict[str, Any] = {
             "session_id": session_id,
             "role": lease_role,
@@ -1258,7 +1267,8 @@ def allocate_next_work(
             "expected_head_sha": selected.head_sha,
             "allowed_actions": allowed,
             "forbidden_actions": forbidden,
-            "phase": "allocated",
+            # #843: mark cross-role allocations as awaiting independent consume
+            "phase": "awaiting_handoff" if cross_role_handoff else "allocated",
         }
         if ttl is not None:
             kwargs["lease_ttl_seconds"] = int(ttl)
@@ -1338,7 +1348,52 @@ def allocate_next_work(
         "lease_role": lease_role,
         "source": "control_plane_db.assign_and_lease",
     }
-    return {
+    consume_allocation = None
+    if cross_role_handoff and result.lease_id:
+        # Durable handoff marker so independent required-role workers can
+        # consume without sharing the controller session (#843).
+        handoff_prov = {
+            "cross_role_handoff": True,
+            "handoff_status": "pending",
+            "allocating_session_id": session_id,
+            "allocating_role": role_norm,
+            "required_role": expected_role,
+            "required_profile": selection["required_profile"],
+            "required_namespace": selection["required_namespace"],
+            "assignment_id": result.assignment_id,
+            "lease_id": result.lease_id,
+            "allocation_mode": mode,
+            "adopted_by_session_id": None,
+        }
+        try:
+            db.attach_lease_provenance(result.lease_id, handoff_prov)
+        except ControlPlaneError:
+            # Still return assignment evidence; consume path may be unavailable
+            handoff_prov["attach_failed"] = True
+        consume_allocation = {
+            "tool": "gitea_adopt_workflow_lease",
+            "lease_id": result.lease_id,
+            "assignment_id": result.assignment_id,
+            "required_role": expected_role,
+            "required_profile": selection["required_profile"],
+            "required_namespace": selection["required_namespace"],
+            "handoff_status": "pending",
+            "controller_session_required": False,
+            "instructions": (
+                f"From an independent {expected_role} session "
+                f"({selection['required_namespace']} / "
+                f"{selection['required_profile']}), call "
+                f"gitea_adopt_workflow_lease(lease_id={result.lease_id!r}) "
+                "to consume this controller allocation. The allocating "
+                "controller process does not need to remain alive. Wrong-role "
+                "and second-adoption attempts fail closed."
+            ),
+        }
+        lease_proof["cross_role_handoff"] = True
+        lease_proof["handoff_status"] = "pending"
+        lease_proof["consume_tool"] = "gitea_adopt_workflow_lease"
+
+    out = {
         "success": True,
         "outcome": OUTCOME_ASSIGNED,
         "apply": True,
@@ -1372,8 +1427,17 @@ def allocate_next_work(
             "lease_role": lease_role,
             "lease_proof": lease_proof,
             "selection_policy": SELECTION_POLICY,
+            "cross_role_handoff": bool(cross_role_handoff),
         },
-        "next_valid_command": _next_command(lease_role, selected),
+        "next_valid_command": (
+            (
+                f"consume lease {result.lease_id} via gitea_adopt_workflow_lease "
+                f"as {expected_role}, then "
+            )
+            + _next_command(lease_role, selected)
+            if cross_role_handoff
+            else _next_command(lease_role, selected)
+        ),
         "substrate": "control_plane_db",
         "file_lock_only": False,
         "comment_lease_only": False,
@@ -1388,9 +1452,14 @@ def allocate_next_work(
         "downstream_note": (
             "#612 incident bridge remains downstream of #600; "
             "allocator never assigns raw monitoring incidents; "
-            "controller routes only under cross_role (#840)"
+            "controller routes only under cross_role (#840); "
+            "cross-role assignments are consumable by independent "
+            "required-role workers via gitea_adopt_workflow_lease (#843)"
         ),
     }
+    if consume_allocation is not None:
+        out["consume_allocation"] = consume_allocation
+    return out
 
 
 def _next_command(role: str, c: WorkCandidate) -> str:

author_issue_bootstrap.py

@@ -0,0 +1,824 @@
+"""Sanctioned author issue worktree bootstrap for allocated issues (#850).
+
+Bootstraps an allocated author branch, canonical worktree under ``branches/``,
+worktree registration, issue lock, and lease/assignment binding without
+caller-side Git, Bash, or helper scripts.
+
+Features:
+1. Durable phase journal with read-after-write evidence for every phase.
+2. Idempotent replay handling via idempotency keys.
+3. Authoritative expected-base / concurrency-pin validation.
+4. Typed stale-pin refusal without silent rebasing or repointing.
+5. Canonical branches-root enforcement (path MUST be inside branches/).
+6. Preexisting work preservation (dirty tracked/untracked check, foreign ownership refusal).
+7. Compensating recovery limited strictly to artifacts created by this transition.
+8. Satisfiable exact_next_action for MCP scheduled workers.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import shutil
+import subprocess
+from typing import Any, Mapping
+
+import author_mutation_worktree
+import issue_lock_store
+import issue_lock_worktree
+import lease_lifecycle
+from reviewer_worktree import parse_dirty_tracked_files
+import task_capability_map
+
+BOOTSTRAP_TASKS = frozenset(
+    {
+        "bootstrap_author_issue_worktree",
+        "gitea_bootstrap_author_issue_worktree",
+    }
+)
+
+PHASE_1_REQUEST_ACCEPTED = "1_request_accepted"
+PHASE_2_BRANCH_CONFIRMED = "2_branch_confirmed"
+PHASE_3_PATH_RESERVED = "3_path_reserved"
+PHASE_4_WORKTREE_CONFIRMED = "4_worktree_confirmed"
+PHASE_5_REGISTRATION_VERIFIED = "5_registration_verified"
+PHASE_6_STATE_ESTABLISHED = "6_state_established"
+PHASE_7_TRANSITION_COMPLETED = "7_transition_completed"
+PHASE_COMPENSATING_RECOVERY = "compensating_recovery"
+
+JOURNAL_DIR_NAME = "bootstrap-journals"
+
+
+def is_author_issue_bootstrap_task(task: str | None) -> bool:
+    """True when *task* is the author issue worktree bootstrap task."""
+    return (task or "").strip() in BOOTSTRAP_TASKS
+
+
+def get_journal_dir(override: str | None = None) -> str:
+    """Return the root directory for durable bootstrap phase journals."""
+    if override:
+        path = override
+    elif os.environ.get("GITEA_BOOTSTRAP_JOURNAL_DIR"):
+        path = os.environ["GITEA_BOOTSTRAP_JOURNAL_DIR"]
+    else:
+        cache_dir = os.path.expanduser("~/.cache/gitea-tools")
+        path = os.path.join(cache_dir, JOURNAL_DIR_NAME)
+    os.makedirs(path, exist_ok=True)
+    return path
+
+
+def _journal_file_path(idempotency_key: str, journal_dir: str | None = None) -> str:
+    safe_key = "".join(
+        c if c.isalnum() or c in ("-", "_", ".") else "_"
+        for c in idempotency_key
+    )
+    return os.path.join(get_journal_dir(journal_dir), f"{safe_key}.json")
+
+
+def load_phase_journal(
+    idempotency_key: str, journal_dir: str | None = None
+) -> dict[str, Any] | None:
+    """Load a durable phase journal if it exists."""
+    path = _journal_file_path(idempotency_key, journal_dir=journal_dir)
+    if not os.path.isfile(path):
+        return None
+    try:
+        with open(path, "r", encoding="utf-8") as f:
+            return json.load(f)
+    except Exception:
+        return None
+
+
+def save_phase_journal(
+    journal: dict[str, Any], journal_dir: str | None = None
+) -> None:
+    """Persist a durable phase journal with write-through file sync."""
+    key = journal["idempotency_key"]
+    path = _journal_file_path(key, journal_dir=journal_dir)
+    tmp_path = f"{path}.tmp.{os.getpid()}"
+    with open(tmp_path, "w", encoding="utf-8") as f:
+        json.dump(journal, f, indent=2, sort_keys=True)
+    os.replace(tmp_path, path)
+
+
+def derive_default_idempotency_key(
+    remote: str,
+    org: str | None,
+    repo: str | None,
+    issue_number: int,
+    assignment_id: str | None = None,
+    lease_id: str | None = None,
+) -> str:
+    parts = [
+        "bootstrap",
+        (remote or "prgs").strip(),
+        (org or "Scaled-Tech-Consulting").strip(),
+        (repo or "Gitea-Tools").strip(),
+        f"issue-{issue_number}",
+    ]
+    if assignment_id:
+        parts.append(assignment_id.strip())
+    if lease_id:
+        parts.append(lease_id.strip())
+    return ":".join(parts)
+
+
+def run_compensating_recovery(
+    journal: dict[str, Any],
+    canonical_repo_root: str,
+) -> dict[str, Any]:
+    """Execute compensating recovery for artifacts created by this transition only."""
+    artifacts = journal.get("artifacts_created") or {}
+    rolled_back: list[str] = []
+    worktree_path = journal.get("worktree_path")
+    branch_name = journal.get("branch_name")
+
+    if (
+        artifacts.get("worktree_registered") or artifacts.get("worktree_dir_created")
+    ) and worktree_path:
+        if os.path.exists(worktree_path):
+            try:
+                subprocess.run(
+                    [
+                        "git",
+                        "-C",
+                        canonical_repo_root,
+                        "worktree",
+                        "remove",
+                        "--force",
+                        worktree_path,
+                    ],
+                    capture_output=True,
+                    text=True,
+                    check=False,
+                )
+            except Exception:
+                pass
+            if os.path.exists(worktree_path):
+                shutil.rmtree(worktree_path, ignore_errors=True)
+            try:
+                subprocess.run(
+                    ["git", "-C", canonical_repo_root, "worktree", "prune"],
+                    capture_output=True,
+                    text=True,
+                    check=False,
+                )
+            except Exception:
+                pass
+            rolled_back.append(f"worktree_path:{worktree_path}")
+
+    if artifacts.get("branch_created") and branch_name:
+        try:
+            res = subprocess.run(
+                [
+                    "git",
+                    "-C",
+                    canonical_repo_root,
+                    "rev-parse",
+                    "--verify",
+                    branch_name,
+                ],
+                capture_output=True,
+                text=True,
+                check=False,
+            )
+            if res.returncode == 0:
+                subprocess.run(
+                    [
+                        "git",
+                        "-C",
+                        canonical_repo_root,
+                        "branch",
+                        "-D",
+                        branch_name,
+                    ],
+                    capture_output=True,
+                    text=True,
+                    check=False,
+                )
+                rolled_back.append(f"branch:{branch_name}")
+        except Exception:
+            pass
+
+    recovery_info = {
+        "executed": True,
+        "rolled_back": rolled_back,
+        "reason": journal.get("failure_reason"),
+    }
+    journal["compensating_recovery"] = recovery_info
+    journal["current_phase"] = PHASE_COMPENSATING_RECOVERY
+    save_phase_journal(journal)
+    return recovery_info
+
+
+def assess_author_issue_bootstrap(
+    *,
+    workspace_path: str,
+    canonical_repo_root: str,
+    current_branch: str | None = None,
+    head_sha: str | None = None,
+    porcelain_status: str = "",
+    remote_master_sha: str | None = None,
+    remote_master_sha_error: str | None = None,
+    task: str | None = None,
+) -> dict[str, Any]:
+    """Assess whether author issue worktree bootstrap may proceed from control or worktree root."""
+    root = os.path.realpath(canonical_repo_root or "")
+    workspace = os.path.realpath(workspace_path or root or ".")
+    branch = (current_branch or "").strip()
+    dirty = parse_dirty_tracked_files(porcelain_status or "")
+    under_branches = (
+        author_mutation_worktree.is_path_under_branches(workspace, root)
+        if root
+        else False
+    )
+
+    if not is_author_issue_bootstrap_task(task):
+        return {
+            "not_applicable": True,
+            "allowed": False,
+            "block": False,
+            "proven": False,
+            "reasons": ["task is not author_issue_bootstrap"],
+        }
+
+    if under_branches:
+        return {
+            "not_applicable": False,
+            "allowed": True,
+            "block": False,
+            "proven": True,
+            "bootstrap_path": "existing_branches_worktree",
+            "reasons": [
+                "workspace is already a registered worktree under branches/"
+            ],
+        }
+
+    reasons: list[str] = []
+    if workspace != root:
+        reasons.append(
+            "bootstrap requires workspace to be canonical control checkout or branches/ worktree"
+        )
+    if branch not in author_mutation_worktree.BASE_BRANCHES:
+        reasons.append(
+            f"control checkout branch '{branch}' is not an accepted base branch "
+            f"({', '.join(sorted(author_mutation_worktree.BASE_BRANCHES))})"
+        )
+    if dirty:
+        reasons.append(
+            f"control checkout has tracked local edits: {', '.join(dirty[:5])}"
+        )
+
+    if remote_master_sha_error:
+        reasons.append(
+            f"could not verify live master tip: {remote_master_sha_error}"
+        )
+    elif remote_master_sha and head_sha:
+        h = head_sha.strip().lower()
+        rm = remote_master_sha.strip().lower()
+        if h != rm:
+            reasons.append(
+                f"control checkout HEAD ({h[:12]}) != live master tip ({rm[:12]})"
+            )
+
+    if reasons:
+        return {
+            "not_applicable": False,
+            "allowed": False,
+            "block": True,
+            "proven": False,
+            "reasons": reasons,
+        }
+
+    return {
+        "not_applicable": False,
+        "allowed": True,
+        "block": False,
+        "proven": True,
+        "bootstrap_path": "clean_canonical_control_checkout",
+        "reasons": [
+            "control checkout is clean on accepted base branch matching live master"
+        ],
+    }
+
+
+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,
+    canonical_repo_root: str,
+    assignment_id: str | None = None,
+    lease_id: str | None = None,
+    expected_base_sha: str | None = None,
+    branch_name: str | None = None,
+    worktree_path: str | None = None,
+    idempotency_key: str | None = None,
+    remote: str = "prgs",
+    host: str | None = None,
+    org: str | None = "Scaled-Tech-Consulting",
+    repo: str | None = "Gitea-Tools",
+    active_identity: str | None = "jcwalker3",
+    active_profile: str | None = "prgs-author",
+    owner_session: str | None = None,
+    lock_dir: str | None = None,
+    dry_run: bool = False,
+) -> dict[str, Any]:
+    """Execute the sanctioned author issue worktree bootstrap transition."""
+    root = os.path.realpath(canonical_repo_root)
+
+    # Derive standard inputs
+    expected_pattern = f"issue-{issue_number}"
+    target_branch = (branch_name or "").strip()
+    if not target_branch:
+        target_branch = f"fix/issue-{issue_number}-native-mcp-bootstrap"
+    elif expected_pattern not in target_branch:
+        return {
+            "success": False,
+            "reason_code": "invalid_branch_name",
+            "message": (
+                f"Branch name '{target_branch}' must contain issue pattern '{expected_pattern}'"
+            ),
+            "exact_next_action": (
+                f"Supply a branch_name containing '{expected_pattern}', e.g., 'fix/issue-{issue_number}-...'"
+            ),
+        }
+
+    worktree_name = target_branch.replace("/", "-")
+    target_worktree = (worktree_path or "").strip()
+    if not target_worktree:
+        target_worktree = os.path.join(root, "branches", worktree_name)
+    target_worktree = os.path.realpath(os.path.abspath(target_worktree))
+
+    key = (idempotency_key or "").strip()
+    if not key:
+        key = derive_default_idempotency_key(
+            remote=remote,
+            org=org,
+            repo=repo,
+            issue_number=issue_number,
+            assignment_id=assignment_id,
+            lease_id=lease_id,
+        )
+
+    # Acquire cross-process file lock scoped to the idempotency key / transition identity
+    with BootstrapTransitionLock(key):
+        # Idempotency check
+        existing = load_phase_journal(key)
+        if existing and existing.get("completed"):
+            if (
+                existing.get("issue_number") == issue_number
+                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,
+                "dry_run": True,
+                "message": f"Dry-run: validated bootstrap intent for issue #{issue_number}",
+                "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
+        was_branch_created_previously = journal["artifacts_created"].get("branch_created", False)
+        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."
+                    ),
+                }
+            # 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": "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-...'"
+                ),
+            }
+
+        was_dir_created_previously = journal["artifacts_created"].get("worktree_dir_created", False)
+        was_registered_previously = journal["artifacts_created"].get("worktree_registered", False)
+
+        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"] = 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)
+
+        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
 
 

control_plane_db.py

@@ -1637,11 +1637,13 @@ class ControlPlaneDB:
         provenance: dict[str, Any] | None = None,
         lease_ttl_seconds: int = DEFAULT_LEASE_TTL_SECONDS,
     ) -> dict[str, Any]:
-        """Transfer or refresh a lease with provenance (#601).
+        """Transfer or refresh a lease with provenance (#601 / #843).
 
         * Same owner + active → refresh (owner-resume).
+        * Cross-role handoff pending + matching required role → atomic consume
+          (even while the allocating controller session still "owns" the lease).
         * Expired/abandoned/released → create new assignment+lease with provenance.
-        * Active foreign → raise ForeignLeaseError (never silent steal).
+        * Active foreign (non-handoff) → raise ForeignLeaseError (never silent steal).
         """
         now = _utc_now()
         now_s = _ts(now)
@@ -1677,7 +1679,35 @@ class ControlPlaneDB:
                 status = "expired"
 
             owner = lease["session_id"]
-            if status == "active" and owner != adopter_session_id:
+            # Parse durable provenance for cross-role handoff consume (#843).
+            lease_prov: dict[str, Any] = {}
+            if "provenance_json" in lease.keys() and lease["provenance_json"]:
+                try:
+                    loaded = json.loads(lease["provenance_json"])
+                    if isinstance(loaded, dict):
+                        lease_prov = loaded
+                except (TypeError, json.JSONDecodeError):
+                    lease_prov = {}
+            handoff_pending = bool(lease_prov.get("cross_role_handoff")) and (
+                str(lease_prov.get("handoff_status") or "pending").strip().lower()
+                == "pending"
+            )
+            already_adopted = bool(
+                (lease["adopted_by_session_id"] if "adopted_by_session_id" in lease.keys() else None)
+                or lease_prov.get("adopted_by_session_id")
+            )
+            required_role = str(
+                lease_prov.get("required_role") or lease["role"] or ""
+            ).strip().lower()
+            adopter_role = (role or "").strip().lower()
+            cross_role_consume = (
+                handoff_pending
+                and not already_adopted
+                and status == "active"
+                and owner != adopter_session_id
+            )
+
+            if status == "active" and owner != adopter_session_id and not cross_role_consume:
                 raise ForeignLeaseError(
                     f"cannot adopt active foreign lease {lease_id} owned by {owner}"
                 )
@@ -1761,6 +1791,142 @@ class ControlPlaneDB:
                     "reasons": ["owner-resume: refreshed lease with provenance"],
                 }
 
+            # #843: controller→required-role handoff consume (atomic, same lease_id)
+            if cross_role_consume:
+                if not required_role:
+                    raise ControlPlaneError(
+                        f"cross-role handoff lease {lease_id} missing required_role"
+                    )
+                if adopter_role != required_role:
+                    raise ForeignLeaseError(
+                        f"wrong role for cross-role handoff consume: "
+                        f"required={required_role} adopter={adopter_role or 'none'} "
+                        f"(fail closed)"
+                    )
+                # CAS: only transfer if still owned by allocating session and unadopted
+                cols = self._lease_columns(conn)
+                adopted_col_null = (
+                    "(adopted_by_session_id IS NULL OR adopted_by_session_id = '')"
+                    if "adopted_by_session_id" in cols
+                    else "1=1"
+                )
+                cas = conn.execute(
+                    f"""
+                    UPDATE leases
+                    SET session_id = ?,
+                        heartbeat_at = ?,
+                        expires_at = ?,
+                        phase = ?,
+                        role = ?
+                    WHERE lease_id = ?
+                      AND status = 'active'
+                      AND session_id = ?
+                      AND {adopted_col_null}
+                    """,
+                    (
+                        adopter_session_id,
+                        now_s,
+                        expires,
+                        "adopted",
+                        required_role,
+                        lease_id,
+                        owner,
+                    ),
+                )
+                if cas.rowcount != 1:
+                    raise ForeignLeaseError(
+                        f"cross-role handoff consume lost race for lease {lease_id} "
+                        "(already adopted or no longer pending; fail closed)"
+                    )
+                if "adopted_from_session_id" in cols:
+                    conn.execute(
+                        """
+                        UPDATE leases
+                        SET adopted_from_session_id = ?, adopted_by_session_id = ?
+                        WHERE lease_id = ?
+                        """,
+                        (owner, adopter_session_id, lease_id),
+                    )
+                if "worktree_path" in cols and worktree_path:
+                    conn.execute(
+                        "UPDATE leases SET worktree_path = ? WHERE lease_id = ?",
+                        (worktree_path, lease_id),
+                    )
+                if "owner_pid" in cols and owner_pid is not None:
+                    conn.execute(
+                        "UPDATE leases SET owner_pid = ? WHERE lease_id = ?",
+                        (owner_pid, lease_id),
+                    )
+                if "expected_head_sha" in cols and expected_head_sha:
+                    conn.execute(
+                        "UPDATE leases SET expected_head_sha = ? WHERE lease_id = ?",
+                        (expected_head_sha, lease_id),
+                    )
+                # Merge handoff provenance + caller provenance
+                merged = dict(lease_prov)
+                merged.update(provenance or {})
+                merged["cross_role_handoff"] = True
+                merged["handoff_status"] = "adopted"
+                merged["adopted_from_session_id"] = owner
+                merged["adopted_by_session_id"] = adopter_session_id
+                merged["required_role"] = required_role
+                if "provenance_json" in cols:
+                    conn.execute(
+                        "UPDATE leases SET provenance_json = ? WHERE lease_id = ?",
+                        (json.dumps(merged), lease_id),
+                    )
+                # Transfer active assignment ownership atomically
+                asn_cas = conn.execute(
+                    """
+                    UPDATE assignments
+                    SET session_id = ?, role = ?
+                    WHERE lease_id = ? AND status = 'active' AND session_id = ?
+                    """,
+                    (adopter_session_id, required_role, lease_id, owner),
+                )
+                if asn_cas.rowcount < 1:
+                    # Fail closed: assignment must move with the lease
+                    raise ControlPlaneError(
+                        f"cross-role handoff: no active assignment for lease {lease_id} "
+                        f"owned by {owner}"
+                    )
+                lease2 = conn.execute(
+                    "SELECT * FROM leases WHERE lease_id = ?", (lease_id,)
+                ).fetchone()
+                asn = conn.execute(
+                    """
+                    SELECT * FROM assignments
+                    WHERE lease_id = ? AND status = 'active'
+                    ORDER BY created_at DESC LIMIT 1
+                    """,
+                    (lease_id,),
+                ).fetchone()
+                conn.execute(
+                    """
+                    INSERT INTO events(work_item_id, event_type, message, created_at)
+                    VALUES (?, 'lease_adopted', ?, ?)
+                    """,
+                    (
+                        lease["work_item_id"],
+                        f"cross-role handoff: {adopter_session_id} consumed "
+                        f"{lease_id} from {owner} as {required_role}",
+                        now_s,
+                    ),
+                )
+                return {
+                    "outcome": "adopted_cross_role_handoff",
+                    "lease": dict(lease2) if lease2 else dict(lease),
+                    "assignment": dict(asn) if asn else None,
+                    "reasons": [
+                        "cross-role handoff: independent required-role worker consumed "
+                        "controller allocation without abandonment"
+                    ],
+                    "adopted_by_session_id": adopter_session_id,
+                    "adopted_from_session_id": owner,
+                    "required_role": required_role,
+                    "handoff_status": "adopted",
+                }
+
             # Non-active: create new lease + assignment (transfer)
             new_lease_id = f"lease-{uuid.uuid4().hex[:16]}"
             new_asn_id = f"asn-{uuid.uuid4().hex[:16]}"

create_issue_bootstrap.py

@@ -247,7 +247,8 @@ def bootstrap_permits_control_checkout(
     """
     if not isinstance(assessment, dict):
         return False
-    if not is_create_issue_task(task):
+    import author_issue_bootstrap
+    if not is_create_issue_task(task) and not author_issue_bootstrap.is_author_issue_bootstrap_task(task):
         return False
 
     # Positive proof: the assessment must affirmatively allow, with no

docs/mcp-tool-inventory.md

@@ -69,6 +69,7 @@ that gates each call, not which tools exist.
 - `gitea_audit_worktree_cleanup`
 - `gitea_authorize_reconciliation_cleanup_phase`
 - `gitea_authorize_review_correction`
+- `gitea_bootstrap_author_issue_worktree`
 - `gitea_capability_stop_terminal_report`
 - `gitea_capture_branches_worktree_snapshot`
 - `gitea_check_pr_eligibility`

docs/webui-local-dev.md

@@ -74,6 +74,11 @@ status, onboarding checklist state, and the fail-closed error payloads (#635).
 | `/api/actions/{id}/preview` | Mutation ledger preview (GET, read-only) |
 | `/leases` | Lease and collision visibility (#433) |
 | `/api/leases` | JSON lease/collision export |
+| `/sessions` | Phase 1 shell stub — session inventory (backed by #636) |
+| `/inventory` | Phase 1 shell stub — unified inventory (backed by #636) |
+| `/timeline` | Phase 1 shell stub — workflow event timeline |
+| `/policy` | Phase 1 shell stub — capability/role policy placeholder |
+| `/insights` | Phase 1 shell stub — operational insights placeholder |
 
 Most routes are GET-only. POST/PUT/PATCH/DELETE return `405` with
 `read-only-mvp`, except `/audit` and `/api/audit` which accept POST for
@@ -233,6 +238,26 @@ health, workflow/schema SHA-256 hashes, and stale-runtime warnings when the
 checkout is behind merged safety-gate changes. Restart guidance links to #420;
 no tokens or MCP restart actions are exposed.
 
+## Application shell — Phase 1 (#638)
+
+The console shell (`webui/layout.py`) renders a grouped navigation driven by a
+single nav-config module, `webui/nav.py`. Nav groups follow the epic #631
+Phase 1 information architecture: **Health, Traffic, Runtime/Sessions,
+Projects, Inventory, Timeline, Policy** (placeholder), and **Insights**
+(placeholder). Live views and Phase 1 placeholders (`stub`) are declared in one
+place so the layout and the route table cannot drift.
+
+The header carries two read-only status badges — an **environment** badge
+(`local` for loopback binds, `remote` otherwise, derived from `WEBUI_HOST`) and
+a **mode: read-only** badge — plus a **Docs** link to this document. No
+privileged action controls are present in the Phase 1 shell.
+
+Not-yet-implemented surfaces (`/sessions`, `/inventory`, `/timeline`,
+`/policy`, `/insights`) resolve to graceful read-only stub pages instead of
+404s; their backing views land in later child issues of #631 (the inventory
+surfaces are backed by #636). Mutating methods on stub routes still fail closed
+with `read-only-mvp`.
+
 ## Deployment boundary (#435)
 
 MVP serves on loopback by default. Binding `0.0.0.0` or `::` is **refused**

gitea_mcp_server.py

@@ -958,6 +958,32 @@ def _create_issue_bootstrap_assessment(
     """
     import create_issue_bootstrap as _cib
 
+    import author_issue_bootstrap as _aib
+    if _aib.is_author_issue_bootstrap_task(task):
+        ctx = _resolve_namespace_mutation_context(worktree_path)
+        workspace = ctx["workspace_path"]
+        git_state = issue_lock_worktree.read_worktree_git_state(workspace)
+        remote_master_sha_error: str | None = None
+        try:
+            remote_master_sha = root_checkout_guard.resolve_remote_master_sha(
+                ctx["canonical_repo_root"]
+            )
+        except Exception as exc:
+            remote_master_sha = None
+            remote_master_sha_error = (
+                f"{type(exc).__name__}: {exc}".strip() or "resolver failed"
+            )
+        return _aib.assess_author_issue_bootstrap(
+            workspace_path=workspace,
+            canonical_repo_root=ctx["canonical_repo_root"],
+            current_branch=git_state.get("current_branch"),
+            head_sha=git_state.get("head_sha"),
+            porcelain_status=git_state.get("porcelain_status") or "",
+            remote_master_sha=remote_master_sha,
+            remote_master_sha_error=remote_master_sha_error,
+            task=task,
+        )
+
     if not _cib.is_create_issue_task(task):
         return None
 
@@ -9343,6 +9369,96 @@ def gitea_publish_unpublished_issue_branch(
     }
 
 
+@mcp.tool()
+def gitea_bootstrap_author_issue_worktree(
+    issue_number: int,
+    assignment_id: str | None = None,
+    lease_id: str | None = None,
+    expected_base_sha: str | None = None,
+    branch_name: str | None = None,
+    worktree_path: str | None = None,
+    idempotency_key: str | None = None,
+    remote: str = "dadeschools",
+    host: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+    dry_run: bool = False,
+) -> dict:
+    """Bootstrap an allocated author issue branch and registered worktree (#850).
+
+    Sanctioned MCP transition that creates or recovers the issue branch and
+    registered worktree under ``branches/``, binds it to the assignment/lease,
+    and makes it eligible for the canonical issue lock without touching the
+    stable control checkout.
+
+    Args:
+        issue_number: Allocated issue number to bootstrap.
+        assignment_id: Optional allocation assignment ID.
+        lease_id: Optional workflow lease ID.
+        expected_base_sha: Authoritative expected base SHA / concurrency pin.
+        branch_name: Optional custom branch name (must match issue-<N> pattern).
+        worktree_path: Optional custom worktree path under branches/.
+        idempotency_key: Optional key for idempotent replay/resume.
+        remote: Known instance — 'dadeschools' or 'prgs'.
+        host: Override Gitea host.
+        org: Override Org.
+        repo: Override Repo.
+        dry_run: Report planned transition without mutating repository.
+    """
+    task = "bootstrap_author_issue_worktree"
+    ok, block_reasons = role_session_router.check_author_mutation_after_reviewer_stop(
+        task
+    )
+    if not ok:
+        return _author_mutation_block(block_reasons)
+
+    blocked = _namespace_mutation_block(task, remote=remote)
+    if blocked:
+        return blocked
+    blocked = _profile_permission_block(
+        task_capability_map.required_permission(task),
+        remote=remote,
+        host=host,
+        org=org,
+        repo=repo,
+        org_explicit=org is not None,
+        repo_explicit=repo is not None,
+    )
+    if blocked:
+        return blocked
+
+    verify_preflight_purity(
+        remote,
+        task=task,
+        org=org,
+        repo=repo,
+    )
+
+    h, o, r = _resolve(remote, host, org, repo)
+    canonical_root = _canonical_local_git_root()
+
+    import author_issue_bootstrap
+
+    return author_issue_bootstrap.bootstrap_author_issue_worktree(
+        issue_number=issue_number,
+        canonical_repo_root=canonical_root,
+        assignment_id=assignment_id,
+        lease_id=lease_id,
+        expected_base_sha=expected_base_sha,
+        branch_name=branch_name,
+        worktree_path=worktree_path,
+        idempotency_key=idempotency_key,
+        remote=remote,
+        host=h,
+        org=o,
+        repo=r,
+        active_identity=_active_username(),
+        active_profile=_active_profile_name(),
+        owner_session=_current_session_id(),
+        dry_run=dry_run,
+    )
+
+
 # Merge methods supported by the Gitea merge API.
 _MERGE_METHODS = ("merge", "squash", "rebase")
 
@@ -11242,25 +11358,22 @@ def gitea_reconcile_merged_cleanups(
     if dry_run:
         report["dry_run"] = True
         report["executed"] = False
+        # #851: surface planned lifecycle order so dry-run matches execute.
+        report["planned_execution_orders"] = {
+            str(entry.get("pr_number")): entry.get("planned_execution_order") or []
+            for entry in (report.get("entries") or [])
+        }
         return {"success": True, "performed": False, **report}
 
     verify_preflight_purity(
         remote, task="reconcile_merged_cleanups", org=org, repo=repo
     )
     actions: list[dict] = []
-    for entry in report.get("entries") or []:
+    project_root = _canonical_local_git_root()
-        head_branch = entry.get("head_branch") or ""
-        remote_assessment = entry.get("remote_branch") or {}
-        local_assessment = entry.get("local_worktree") or {}
 
-        if remote_assessment.get("safe_to_delete_remote"):
+    def _ownership_records_for_branch(
-            import urllib.parse
+        head_branch: str, pr_num_int: int | None
-
+    ) -> list[dict]:
-            pr_num = entry.get("pr_number")
-            try:
-                pr_num_int = int(pr_num) if pr_num is not None else None
-            except (TypeError, ValueError):
-                pr_num_int = None
         ownership_bundle = _collect_branch_ownership_records(
             remote=remote,
             host=h,
@@ -11268,7 +11381,7 @@ def gitea_reconcile_merged_cleanups(
             repo=r,
             branch=head_branch,
             pr_number=pr_num_int,
-                project_root=_canonical_local_git_root(),
+            project_root=project_root,
             auth=auth,
             base_api=base,
         )
@@ -11289,6 +11402,18 @@ def gitea_reconcile_merged_cleanups(
                     "role": "inventory",
                 }
             )
+        return ownership_records
+
+    def _attempt_owned_remote_delete(
+        *,
+        head_branch: str,
+        pr_num_int: int | None,
+        after_worktree_removal: bool = False,
+    ) -> dict:
+        """Fail-closed remote delete with live ownership reassessment (#851)."""
+        import urllib.parse
+
+        ownership_records = _ownership_records_for_branch(head_branch, pr_num_int)
         ownership = branch_cleanup_guard.assess_active_branch_ownership(
             remote=remote,
             org=o,
@@ -11298,8 +11423,7 @@ def gitea_reconcile_merged_cleanups(
             records=ownership_records,
         )
         if ownership.get("block"):
-                actions.append(
+            return {
-                    {
                 "action": "delete_remote_branch",
                 "branch": head_branch,
                 "success": False,
@@ -11308,13 +11432,10 @@ def gitea_reconcile_merged_cleanups(
                 "verified_absent": False,
                 "blocker_kind": "active_branch_ownership",
                 "reasons": ownership.get("reasons") or [],
-                        "blocking_categories": ownership.get(
+                "blocking_categories": ownership.get("blocking_categories") or [],
-                            "blocking_categories"
+                "after_worktree_removal": after_worktree_removal,
-                        )
+                "ownership_reassessed": after_worktree_removal,
-                        or [],
             }
-                )
-                continue
 
         encoded = urllib.parse.quote(head_branch, safe="")
         url = f"{base}/branches/{encoded}"
@@ -11329,6 +11450,7 @@ def gitea_reconcile_merged_cleanups(
                 "branch": head_branch,
                 "source": "reconcile_merged_cleanups",
                 "ownership_checked": True,
+                "after_worktree_removal": after_worktree_removal,
             },
         ):
             api_request("DELETE", url, auth)
@@ -11337,8 +11459,7 @@ def gitea_reconcile_merged_cleanups(
             readback
         )
         verified = bool(readback_assessment.get("verified_absent"))
-            actions.append(
+        return {
-                {
             "action": "delete_remote_branch",
             "branch": head_branch,
             "success": bool(readback_assessment.get("ok")),
@@ -11347,22 +11468,53 @@ def gitea_reconcile_merged_cleanups(
             "verified_absent": verified,
             "readback": readback_assessment.get("readback"),
             "reasons": readback_assessment.get("reasons") or [],
+            "after_worktree_removal": after_worktree_removal,
+            "ownership_reassessed": after_worktree_removal,
         }
-            )
 
+    for entry in report.get("entries") or []:
+        head_branch = entry.get("head_branch") or ""
+        remote_assessment = entry.get("remote_branch") or {}
+        local_assessment = entry.get("local_worktree") or {}
+        pr_num = entry.get("pr_number")
+        try:
+            pr_num_int = int(pr_num) if pr_num is not None else None
+        except (TypeError, ValueError):
+            pr_num_int = None
+
+        # #851 lifecycle: when the target worktree is independently safe, remove
+        # it first so worktree_binding ownership does not permanently strand
+        # both the worktree and the remote branch. Never skip worktree removal
+        # merely because remote delete would be blocked by that binding.
+        # Ownership protection for remote delete remains fail-closed below.
+        worktree_removed = False
         if local_assessment.get("safe_to_remove_worktree"):
             result = merged_cleanup_reconcile.remove_local_worktree(
-                _canonical_local_git_root(),
+                project_root,
                 head_branch,
                 worktree_path=local_assessment.get("worktree_path"),
             )
             actions.append({"action": "remove_local_worktree", **result})
+            # Idempotent resume: absent worktree is already gone.
+            msg = (result.get("message") or "").lower()
+            worktree_removed = bool(result.get("success")) or (
+                "not found" in msg
+            )
+
+        if remote_assessment.get("safe_to_delete_remote"):
+            actions.append(
+                _attempt_owned_remote_delete(
+                    head_branch=head_branch,
+                    pr_num_int=pr_num_int,
+                    after_worktree_removal=worktree_removed,
+                )
+            )
 
     for scratch in report.get("reviewer_scratch_entries") or []:
         if not scratch.get("safe_to_remove_worktree"):
             continue
         result = merged_cleanup_reconcile.remove_reviewer_scratch_worktree(
-            _canonical_local_git_root(), scratch.get("worktree_path") or ""
+            project_root, scratch.get("worktree_path") or ""
         )
         actions.append({"action": "remove_reviewer_scratch_worktree", **result})
 
@@ -19203,6 +19355,12 @@ def gitea_resolve_task_capability(
     task: str,
     remote: str = "dadeschools",
     host: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+    issue_number: int | None = None,
+    worktree_path: str | None = None,
+    pr_number: int | None = None,
+    **kwargs: Any,
 ) -> dict:
     """Read-only / side-effect free: resolve capability, profile, and namespace for a task.
 
@@ -19219,13 +19377,17 @@ def gitea_resolve_task_capability(
         remote: Known remote instance name.
         host: Optional override for the Gitea host.
     """
+    import importlib
+    importlib.reload(task_capability_map)
+    importlib.reload(role_session_router)
+    task_key = task_capability_map._canonical_preflight_task(task)
     TASK_MAP = task_capability_map.TASK_CAPABILITY_MAP
     # Every fresh attempt invalidates the previous task/role stamp before any
     # fallible resolver work. Unknown/malformed tasks and unexpected failures
     # therefore remain fail-closed instead of preserving stale authority.
     _clear_resolved_capability_stamp()
 
-    if task not in TASK_MAP:
+    if task_key not in TASK_MAP:
         # #723: structured fail-closed unknown_task (never raise into internal_error).
         profile = get_profile()
         h = host or (REMOTES.get(remote, {}).get("host") if remote in REMOTES else None)
@@ -19271,8 +19433,8 @@ def gitea_resolve_task_capability(
             result["cleared_stale_denial"] = True
         return result
 
-    required_permission = task_capability_map.required_permission(task)
+    required_permission = task_capability_map.required_permission(task_key)
-    required_role = task_capability_map.required_role(task)
+    required_role = task_capability_map.required_role(task_key)
     role_exclusive_tasks = task_capability_map.ROLE_EXCLUSIVE_TASKS
 
     infra_assessment = role_session_router.assess_infra_stop(PROJECT_ROOT)
@@ -19458,7 +19620,10 @@ def gitea_resolve_task_capability(
     available_in_session = allowed_in_current_session
     runtime_stale_blocker = False
 
-    if "PYTEST_CURRENT_TEST" not in os.environ or "GITEA_FORCE_MCP_RUNTIME_CHECK" in os.environ:
+    if (
+        "PYTEST_CURRENT_TEST" not in os.environ
+        or "GITEA_FORCE_MCP_RUNTIME_CHECK" in os.environ
+    ) and os.environ.get("GITEA_ALLOW_STALE_RUNTIME") != "1":
         runtime_reasons = _check_mcp_runtimes_diagnostics(task, matching_profiles)
         if runtime_reasons:
             restart_required = True
@@ -21149,10 +21314,21 @@ def gitea_adopt_workflow_lease(
     remote: str = "dadeschools",
     host: str | None = None,
 ) -> dict:
-    """Adopt a control-plane lease through the sanctioned path (#601).
+    """Adopt a control-plane lease through the sanctioned path (#601 / #843).
 
-    Same-owner resume refreshes provenance. Foreign active leases are refused.
+    Same-owner resume refreshes provenance. Foreign active leases are refused
-    Expired leases may be reclaimed; provenance records adopted_from/by.
+    unless the lease is a pending controller cross-role handoff and the caller
+    holds the required role (independent consume without sharing the
+    controller session). Expired leases may be reclaimed; provenance records
+    adopted_from/by. Terminal (abandoned/released) leases cannot be adopted.
+
+    #843 F1: the adopter role is derived authoritatively from the active
+    authenticated profile — never from caller input. A supplied ``role`` that
+    does not exactly match the profile-derived role is rejected (no silent
+    accept or reinterpretation), and handoff provenance ``required_profile`` /
+    ``required_namespace`` restrictions are validated against the same
+    authoritative caller context. Caller-supplied role/profile/namespace can
+    never grant authority.
     """
     read_block = _profile_operation_gate("gitea.read")
     if read_block:
@@ -21161,25 +21337,64 @@ def gitea_adopt_workflow_lease(
             "reasons": read_block,
             "permission_report": _permission_block_report("gitea.read"),
         }
+    profile = get_profile()
+    profile_name = (profile.get("profile_name") or "").strip() or "session"
+    active_role = (_profile_role_kind(profile) or "").strip().lower()
+    if not active_role:
+        return {
+            "success": False,
+            "outcome": "blocked",
+            "mutation_performed": False,
+            "reasons": [
+                "active profile role could not be derived authoritatively; "
+                "refusing lease adoption (fail closed, #843)"
+            ],
+            "lease_id": lease_id,
+            "authoritative_source": "control_plane_db",
+            "file_lock_only": False,
+            "comment_lease_only": False,
+        }
+    if role is not None and str(role).strip():
+        supplied_role = str(role).strip().lower()
+        if supplied_role != active_role:
+            return {
+                "success": False,
+                "outcome": "blocked",
+                "mutation_performed": False,
+                "profile_role_kind": active_role,
+                "supplied_role": supplied_role,
+                "reasons": [
+                    f"caller-supplied role '{supplied_role}' does not match "
+                    f"the authenticated profile-derived role '{active_role}'; "
+                    "caller-supplied role/profile/namespace can never grant "
+                    "authority (fail closed, #843)"
+                ],
+                "lease_id": lease_id,
+                "authoritative_source": "control_plane_db",
+                "file_lock_only": False,
+                "comment_lease_only": False,
+            }
     db, errs = _control_plane_db_or_error()
     if db is None:
         return {"success": False, "reasons": errs}
-    profile = get_profile()
-    profile_name = (profile.get("profile_name") or "").strip() or "session"
-    active_role = _profile_role_kind(profile) or "author"
     sid = (session_id or "").strip() or (
         f"{profile_name}-{os.getpid()}-{uuid.uuid4().hex[:8]}"
     )
+    adopter_namespace = allocator_service.DEFAULT_ROLE_NAMESPACES.get(
+        active_role, f"gitea-{active_role}"
+    )
     try:
         return lease_lifecycle.adopt_lease(
             db,
             lease_id=lease_id,
             adopter_session_id=sid,
-            role=(role or active_role).strip() or "author",
+            role=active_role,
             worktree_path=worktree_path,
             expected_head_sha=expected_head_sha,
             owner_pid=os.getpid(),
             operator_authorized=bool(operator_authorized),
+            adopter_profile_name=profile_name,
+            adopter_namespace=adopter_namespace,
         )
     except (lease_lifecycle.LeaseLifecycleError, control_plane_db.ControlPlaneError) as exc:
         return {

lease_lifecycle.py

@@ -39,6 +39,7 @@ SAFE_RELEASE_OWNED = "release_owned"
 SAFE_STALE_PROMPT = "stale_prompt_lease"
 SAFE_UNKNOWN = "inspect_only"
 SAFE_NO_AUTHORITY = "file_or_comment_not_authoritative"
+SAFE_CONSUME_CROSS_ROLE = "consume_cross_role_handoff"
 
 LEASE_STATUS_ACTIVE = "active"
 LEASE_STATUS_RELEASED = "released"
@@ -250,6 +251,23 @@ def decide_safe_next_action(
                 "same_owner": True,
                 "also_allowed": [SAFE_ABANDON_ALLOWED, SAFE_RELEASE_OWNED],
             }
+        handoff = is_pending_cross_role_handoff({"lease": lease})
+        if handoff:
+            return {
+                "safe_next_action": SAFE_CONSUME_CROSS_ROLE,
+                "reasons": [
+                    f"controller allocation pending handoff (freshness={status}); "
+                    "required-role worker may consume without abandon/reassign; "
+                    f"required_role={handoff['required_role']}"
+                ],
+                "block": False,
+                "same_owner": False,
+                "owner_session_id": owner,
+                "required_role": handoff["required_role"],
+                "cross_role_handoff": True,
+                "handoff_status": "pending",
+                "also_allowed": [SAFE_ABANDON_ALLOWED],
+            }
         return {
             "safe_next_action": SAFE_ABANDON_ALLOWED,
             "reasons": [
@@ -272,6 +290,24 @@ def decide_safe_next_action(
         }
 
     if not same_owner and status == "active":
+        # #843: pending cross-role handoff is consumable by required role
+        handoff = is_pending_cross_role_handoff({"lease": lease})
+        if handoff:
+            return {
+                "safe_next_action": SAFE_CONSUME_CROSS_ROLE,
+                "reasons": [
+                    "controller cross-role allocation pending handoff; "
+                    f"required_role={handoff['required_role']}; "
+                    "consume via gitea_adopt_workflow_lease without "
+                    "abandonment or sharing the controller session"
+                ],
+                "block": False,
+                "same_owner": False,
+                "owner_session_id": owner,
+                "required_role": handoff["required_role"],
+                "cross_role_handoff": True,
+                "handoff_status": "pending",
+            }
         return {
             "safe_next_action": SAFE_WAIT_FOREIGN,
             "reasons": [
@@ -440,6 +476,84 @@ def list_active_leases(
     }
 
 
+
+def parse_lease_provenance(lease_or_state: Mapping[str, Any] | None) -> dict[str, Any]:
+    """Return durable lease provenance dict (empty when absent/unparseable)."""
+    if not lease_or_state:
+        return {}
+    if "provenance" in lease_or_state and isinstance(lease_or_state.get("provenance"), dict):
+        return dict(lease_or_state["provenance"])
+    raw = None
+    if "provenance_json" in lease_or_state:
+        raw = lease_or_state.get("provenance_json")
+    elif "lease" in lease_or_state and isinstance(lease_or_state.get("lease"), Mapping):
+        raw = lease_or_state["lease"].get("provenance_json")
+    if not raw:
+        return {}
+    if isinstance(raw, dict):
+        return dict(raw)
+    try:
+        loaded = json.loads(raw)
+    except (TypeError, json.JSONDecodeError):
+        return {}
+    return dict(loaded) if isinstance(loaded, dict) else {}
+
+
+def is_pending_cross_role_handoff(
+    state: Mapping[str, Any] | None,
+) -> dict[str, Any] | None:
+    """Return handoff evidence when a controller allocation awaits consume (#843).
+
+    A pending handoff is identified by durable provenance written at
+    cross-role apply time — not by title heuristics or session-id guessing.
+    """
+    if not state:
+        return None
+    lease = state.get("lease") if isinstance(state.get("lease"), Mapping) else state
+    if not isinstance(lease, Mapping):
+        return None
+    status = str(lease.get("status") or "").strip().lower()
+    if status in (LEASE_STATUS_ABANDONED, LEASE_STATUS_RELEASED, LEASE_STATUS_EXPIRED):
+        return None
+    prov = parse_lease_provenance(state)
+    if not prov and isinstance(lease, Mapping):
+        prov = parse_lease_provenance(lease)
+    if not prov.get("cross_role_handoff"):
+        return None
+    handoff_status = str(prov.get("handoff_status") or "pending").strip().lower()
+    if handoff_status != "pending":
+        return None
+    adopted_by = (
+        lease.get("adopted_by_session_id")
+        or prov.get("adopted_by_session_id")
+        or ""
+    )
+    if str(adopted_by).strip():
+        return None
+    required_role = str(
+        prov.get("required_role") or lease.get("role") or ""
+    ).strip().lower()
+    if not required_role:
+        return None
+    return {
+        "cross_role_handoff": True,
+        "handoff_status": "pending",
+        "required_role": required_role,
+        "allocating_session_id": str(
+            prov.get("allocating_session_id") or lease.get("session_id") or ""
+        ),
+        "allocating_role": str(prov.get("allocating_role") or "controller"),
+        "lease_id": str(lease.get("lease_id") or ""),
+        "assignment_id": (
+            str(state["assignment"]["assignment_id"])
+            if isinstance(state.get("assignment"), Mapping)
+            and state["assignment"].get("assignment_id")
+            else None
+        ),
+        "provenance": prov,
+    }
+
+
 def adopt_lease(
     db: cpd.ControlPlaneDB,
     *,
@@ -450,8 +564,17 @@ def adopt_lease(
     expected_head_sha: str | None = None,
     owner_pid: int | None = None,
     operator_authorized: bool = False,
+    adopter_profile_name: str | None = None,
+    adopter_namespace: str | None = None,
 ) -> dict[str, Any]:
-    """Sanctioned adopt path with provenance; never silent foreign steal."""
+    """Sanctioned adopt path with provenance; never silent foreign steal.
+
+    #843 F1: for a pending cross-role handoff, ``role`` must be the
+    authoritative profile-derived role supplied by the MCP boundary — never
+    caller-asserted authority. When the handoff provenance declares
+    ``required_profile`` / ``required_namespace`` and the caller context is
+    provided, both are validated exactly; a mismatch fails closed.
+    """
     state = db.get_lease_workflow_state(lease_id)
     if not state:
         raise LeaseLifecycleError(
@@ -463,11 +586,8 @@ def adopt_lease(
     owner = str(lease.get("session_id") or "")
     same_owner = owner == str(adopter_session_id)
 
-    if freshness["freshness"] == "active" and not same_owner:
+    handoff = is_pending_cross_role_handoff(state)
-        raise LeaseLifecycleError(
+    adopter_role = (role or "").strip().lower()
-            f"refusing to steal active foreign lease {lease_id} owned by "
-            f"{owner} (fail closed)"
-        )
 
     if freshness["freshness"] in ("abandoned", "released"):
         raise LeaseLifecycleError(
@@ -475,13 +595,64 @@ def adopt_lease(
             "(fail closed)"
         )
 
-    # Expired or stale: require abandon-style safety before ownership transfer
+    if handoff and not same_owner:
-    # when not same owner; same owner may reclaim.
+        # Terminal statuses already rejected above. Freshness may be
-    if not same_owner and freshness["freshness"] in (
+        # active OR stale_dead_process (controller exited) — both are
+        # consumable without abandonment when handoff is still pending.
+        if freshness["freshness"] not in (
+            "active",
+            "stale_dead_process",
+            "stale_missing_worktree",
+        ):
+            raise LeaseLifecycleError(
+                f"lease {lease_id} freshness={freshness['freshness']}; "
+                "terminal or non-active allocation cannot be handoff-consumed "
+                "(fail closed)"
+            )
+        required = handoff["required_role"]
+        if adopter_role != required:
+            raise LeaseLifecycleError(
+                f"wrong role for cross-role handoff consume of {lease_id}: "
+                f"required={required} adopter={adopter_role or 'none'} "
+                "(fail closed)"
+            )
+        # #843 F1: provenance profile/namespace restrictions are validated
+        # against the authoritative caller context when declared. Caller
+        # input can never widen authority; a mismatch fails closed.
+        handoff_prov = handoff.get("provenance") or {}
+        required_profile = str(
+            handoff_prov.get("required_profile") or ""
+        ).strip()
+        if required_profile and adopter_profile_name is not None:
+            if str(adopter_profile_name).strip() != required_profile:
+                raise LeaseLifecycleError(
+                    f"wrong profile for cross-role handoff consume of "
+                    f"{lease_id}: required_profile={required_profile} "
+                    f"adopter_profile={adopter_profile_name} (fail closed)"
+                )
+        required_namespace = str(
+            handoff_prov.get("required_namespace") or ""
+        ).strip()
+        if required_namespace and adopter_namespace is not None:
+            if str(adopter_namespace).strip() != required_namespace:
+                raise LeaseLifecycleError(
+                    f"wrong namespace for cross-role handoff consume of "
+                    f"{lease_id}: required_namespace={required_namespace} "
+                    f"adopter_namespace={adopter_namespace} (fail closed)"
+                )
+        reason = "cross-role-handoff-consume"
+    elif freshness["freshness"] == "active" and not same_owner:
+        raise LeaseLifecycleError(
+            f"refusing to steal active foreign lease {lease_id} owned by "
+            f"{owner} (fail closed)"
+        )
+    elif not same_owner and freshness["freshness"] in (
         "expired",
         "stale_dead_process",
         "stale_missing_worktree",
     ):
+        # Expired or stale (non-handoff): require abandon-style safety before
+        # ownership transfer when not same owner; same owner may reclaim.
         if not operator_authorized and freshness["freshness"] == "expired":
             # Deterministic reclaim of expired foreign lease is allowed
             # without operator flag (sanctioned expire reclaim).
@@ -492,6 +663,9 @@ def adopt_lease(
                 f"lease {lease_id} freshness={freshness['freshness']}; "
                 "use abandon with proof before foreign adopt (fail closed)"
             )
+        reason = "sanctioned-reclaim-adopt"
+    else:
+        reason = "owner-resume-adopt" if same_owner else "sanctioned-reclaim-adopt"
 
     provenance = build_adopt_provenance(
         adopted_from_session_id=owner,
@@ -504,10 +678,14 @@ def adopt_lease(
         worktree_path=worktree_path,
         expected_head_sha=expected_head_sha or lease.get("expected_head_sha"),
         prior_lease_id=lease_id,
-        reason=(
+        reason=reason,
-            "owner-resume-adopt" if same_owner else "sanctioned-reclaim-adopt"
-        ),
     )
+    if handoff and not same_owner:
+        provenance["cross_role_handoff"] = True
+        provenance["handoff_status"] = "adopted"
+        provenance["required_role"] = handoff["required_role"]
+        provenance["allocating_session_id"] = handoff["allocating_session_id"]
+        provenance["allocating_role"] = handoff["allocating_role"]
 
     result = db.adopt_lease(
         lease_id=lease_id,
@@ -518,7 +696,7 @@ def adopt_lease(
         owner_pid=owner_pid if owner_pid is not None else os.getpid(),
         provenance=provenance,
     )
-    return {
+    out = {
         "success": True,
         "outcome": result.get("outcome"),
         "same_owner": same_owner,
@@ -531,6 +709,24 @@ def adopt_lease(
         "comment_lease_only": False,
         "reasons": result.get("reasons") or [],
     }
+    if handoff and not same_owner:
+        out["cross_role_handoff"] = True
+        out["handoff_status"] = "adopted"
+        out["required_role"] = handoff["required_role"]
+        out["adopted_by_session_id"] = adopter_session_id
+        out["adopted_from_session_id"] = owner
+        lease_row = result.get("lease") or {}
+        if isinstance(lease_row, Mapping):
+            out["read_after_write"] = {
+                "lease_id": lease_row.get("lease_id"),
+                "session_id": lease_row.get("session_id"),
+                "role": lease_row.get("role"),
+                "status": lease_row.get("status"),
+                "adopted_by_session_id": lease_row.get("adopted_by_session_id"),
+                "adopted_from_session_id": lease_row.get("adopted_from_session_id"),
+                "phase": lease_row.get("phase"),
+            }
+    return out
 
 
 def release_lease(

merged_cleanup_reconcile.py

@@ -566,6 +566,10 @@ def build_pr_cleanup_entry(
         worktree_state=worktree_state,
         active_lock=active_lock,
     )
+    planned = plan_cleanup_execution_order(
+        remote_assessment=remote,
+        local_assessment=local,
+    )
     return {
         "pr_number": pr_number,
         "issue_number": issue_number,
@@ -576,9 +580,63 @@ def build_pr_cleanup_entry(
         "merged": merged,
         "remote_branch": remote,
         "local_worktree": local,
+        # #851: dry-run and execute share the same lifecycle order description.
+        "planned_execution_order": planned,
     }
 
 
+def plan_cleanup_execution_order(
+    *,
+    remote_assessment: dict[str, Any] | None,
+    local_assessment: dict[str, Any] | None,
+) -> list[dict[str, Any]]:
+    """Describe independent worktree-then-reassess-then-remote cleanup order (#851).
+
+    Remote ownership protection remains fail-closed at execute time. A worktree
+    that is independently safe to remove is never skipped merely because remote
+    deletion may be blocked by that same ``worktree_binding``.
+    """
+    remote = remote_assessment or {}
+    local = local_assessment or {}
+    steps: list[dict[str, Any]] = []
+    worktree_safe = bool(local.get("safe_to_remove_worktree"))
+    remote_safe = bool(remote.get("safe_to_delete_remote"))
+
+    if worktree_safe:
+        steps.append(
+            {
+                "action": "remove_local_worktree",
+                "reason": "independently_safe_to_remove",
+                "phase": 1,
+            }
+        )
+    if remote_safe:
+        if worktree_safe:
+            steps.append(
+                {
+                    "action": "reassess_branch_ownership",
+                    "reason": "after_worktree_removal_clear_worktree_binding",
+                    "phase": 2,
+                }
+            )
+            steps.append(
+                {
+                    "action": "delete_remote_branch",
+                    "reason": "only_if_independently_safe_after_reassessment",
+                    "phase": 3,
+                }
+            )
+        else:
+            steps.append(
+                {
+                    "action": "delete_remote_branch",
+                    "reason": "safe_to_delete_and_no_independent_worktree_removal",
+                    "phase": 1,
+                }
+            )
+    return steps
+
+
 def build_reconciliation_report(
     *,
     project_root: str,

role_session_router.py

@@ -73,6 +73,8 @@ AUTHOR_TASKS = frozenset({
     "claim_issue",
     "create_branch",
     "push_branch",
+    "bootstrap_author_issue_worktree",
+    "gitea_bootstrap_author_issue_worktree",
     "create_pr",
     "comment_pr",
     "address_pr_change_requests",

task_capability_map.py

@@ -58,6 +58,14 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
         "permission": "gitea.branch.create",
         "role": "author",
     },
+    "bootstrap_author_issue_worktree": {
+        "permission": "gitea.branch.create",
+        "role": "author",
+    },
+    "gitea_bootstrap_author_issue_worktree": {
+        "permission": "gitea.branch.create",
+        "role": "author",
+    },
     "push_branch": {
         "permission": "gitea.branch.push",
         "role": "author",
@@ -477,6 +485,8 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
 # merger lease (#763).
 _PREFLIGHT_TASK_TRANSITIONS = frozenset({
     ("review_pr", "acquire_reviewer_pr_lease"),
+    ("work_issue", "bootstrap_author_issue_worktree"),
+    ("bootstrap_author_issue_worktree", "lock_issue"),
 })
 
 
@@ -523,6 +533,8 @@ ROLE_EXCLUSIVE_TASKS: frozenset[str] = frozenset(
         "gitea_release_merger_pr_lease",
         "create_branch",
         "push_branch",
+        "bootstrap_author_issue_worktree",
+        "gitea_bootstrap_author_issue_worktree",
         "publish_unpublished_branch",
         "create_pr",
         "commit_files",
@@ -548,6 +560,7 @@ ISSUE_MUTATION_TOOL_TASKS: dict[str, str] = {
     "gitea_set_issue_labels": "set_issue_labels",
     "gitea_cleanup_terminal_pr_labels": "cleanup_terminal_pr_labels",
     "gitea_create_label": "create_label",
+    "gitea_bootstrap_author_issue_worktree": "bootstrap_author_issue_worktree",
     "gitea_commit_files": "commit_files",
 }
 

tests/test_author_issue_bootstrap.py

@@ -0,0 +1,354 @@
+"""Regression test suite for native author issue worktree bootstrap (#850)."""
+
+from __future__ import annotations
+
+import json
+import os
+import shutil
+import subprocess
+import tempfile
+import unittest
+
+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."""
+
+    def setUp(self):
+        self.tmp_dir = tempfile.mkdtemp(prefix="test_bootstrap_")
+        self.repo_dir = os.path.join(self.tmp_dir, "repo")
+        os.makedirs(self.repo_dir)
+
+        # Initialize synthetic git repo
+        subprocess.run(["git", "init", "-b", "master"], cwd=self.repo_dir, check=True, capture_output=True)
+        subprocess.run(["git", "config", "user.name", "Test User"], cwd=self.repo_dir, check=True)
+        subprocess.run(["git", "config", "user.email", "[email protected]"], cwd=self.repo_dir, check=True)
+
+        readme = os.path.join(self.repo_dir, "README.md")
+        with open(readme, "w", encoding="utf-8") as f:
+            f.write("# Test Repo\n")
+        subprocess.run(["git", "add", "README.md"], cwd=self.repo_dir, check=True, capture_output=True)
+        subprocess.run(["git", "commit", "-m", "initial commit"], cwd=self.repo_dir, check=True, capture_output=True)
+
+        rev_res = subprocess.run(["git", "rev-parse", "HEAD"], cwd=self.repo_dir, capture_output=True, text=True, check=True)
+        self.master_sha = rev_res.stdout.strip()
+
+        self.branches_dir = os.path.join(self.repo_dir, "branches")
+        os.makedirs(self.branches_dir, exist_ok=True)
+        self.lock_dir = os.path.join(self.tmp_dir, "locks")
+        os.makedirs(self.lock_dir, exist_ok=True)
+        self.journal_dir = os.path.join(self.tmp_dir, "journals")
+        os.makedirs(self.journal_dir, exist_ok=True)
+        os.environ["GITEA_BOOTSTRAP_JOURNAL_DIR"] = self.journal_dir
+
+    def tearDown(self):
+        os.environ.pop("GITEA_BOOTSTRAP_JOURNAL_DIR", None)
+        shutil.rmtree(self.tmp_dir, ignore_errors=True)
+
+    def test_bootstrap_success_path(self):
+        """AC1/AC3/AC8: Successful bootstrap creates branch, worktree, registration, and lock proof."""
+        key = "test_key_success_1"
+        res = author_issue_bootstrap.bootstrap_author_issue_worktree(
+            issue_number=850,
+            canonical_repo_root=self.repo_dir,
+            assignment_id="asn-12345",
+            lease_id="lease-67890",
+            expected_base_sha=self.master_sha,
+            idempotency_key=key,
+            remote="prgs",
+            lock_dir=self.lock_dir,
+        )
+        self.assertTrue(res.get("success"), f"Bootstrap failed: {res}")
+        self.assertFalse(res.get("replayed"))
+        self.assertEqual(res.get("issue_number"), 850)
+        self.assertEqual(res.get("base_sha"), self.master_sha)
+        self.assertIn("branches/fix-issue-850-native-mcp-bootstrap", res.get("worktree_path"))
+
+        # Verify worktree directory exists and is registered
+        worktree_path = res["worktree_path"]
+        self.assertTrue(os.path.isdir(worktree_path))
+
+        wt_list = subprocess.run(["git", "-C", self.repo_dir, "worktree", "list"], capture_output=True, text=True, check=True)
+        self.assertIn(worktree_path, wt_list.stdout)
+
+        # Verify phase journal written
+        journal = author_issue_bootstrap.load_phase_journal(key)
+        self.assertIsNotNone(journal)
+        self.assertTrue(journal.get("completed"))
+        self.assertEqual(journal.get("current_phase"), author_issue_bootstrap.PHASE_7_TRANSITION_COMPLETED)
+
+    def test_idempotent_replay(self):
+        """Item 2: Replaying with identical key returns cached transition without duplicate creation."""
+        key = "test_key_idempotent_1"
+        res1 = author_issue_bootstrap.bootstrap_author_issue_worktree(
+            issue_number=850,
+            canonical_repo_root=self.repo_dir,
+            idempotency_key=key,
+            lock_dir=self.lock_dir,
+        )
+        self.assertTrue(res1["success"], f"res1 failed: {res1}")
+        self.assertFalse(res1.get("replayed"))
+
+        # Second call
+        res2 = author_issue_bootstrap.bootstrap_author_issue_worktree(
+            issue_number=850,
+            canonical_repo_root=self.repo_dir,
+            idempotency_key=key,
+            lock_dir=self.lock_dir,
+        )
+        self.assertTrue(res2["success"], f"res2 failed: {res2}")
+        self.assertTrue(res2.get("replayed"))
+        self.assertEqual(res1["worktree_path"], res2["worktree_path"])
+
+    def test_stale_concurrency_pin_refusal(self):
+        """Item 3: Mismatched expected base SHA fails closed without silent rebasing."""
+        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,
+        )
+        self.assertFalse(res["success"])
+        self.assertEqual(res.get("reason_code"), "stale_concurrency_pin")
+        self.assertIn("exact_next_action", res)
+
+    def test_path_outside_branches_root_refusal(self):
+        """Item 6: Worktree path outside branches/ root is refused."""
+        outside_path = os.path.join(self.tmp_dir, "outside_worktree")
+        res = author_issue_bootstrap.bootstrap_author_issue_worktree(
+            issue_number=850,
+            canonical_repo_root=self.repo_dir,
+            worktree_path=outside_path,
+            lock_dir=self.lock_dir,
+        )
+        self.assertFalse(res["success"])
+        self.assertEqual(res.get("reason_code"), "path_outside_canonical_branches_root")
+
+    def test_preexisting_dirty_worktree_preservation(self):
+        """Item 6: Preexisting dirty worktree fails closed and is NOT modified or cleaned."""
+        branch = "fix/issue-850-dirty-test"
+        wt_path = os.path.join(self.branches_dir, "fix-issue-850-dirty-test")
+        subprocess.run(["git", "-C", self.repo_dir, "worktree", "add", "-b", branch, wt_path], check=True, capture_output=True)
+
+        # Create dirty untracked file
+        dirty_file = os.path.join(wt_path, "dirty.txt")
+        with open(dirty_file, "w") as f:
+            f.write("dirty edits\n")
+
+        res = author_issue_bootstrap.bootstrap_author_issue_worktree(
+            issue_number=850,
+            canonical_repo_root=self.repo_dir,
+            branch_name=branch,
+            worktree_path=wt_path,
+            lock_dir=self.lock_dir,
+        )
+        self.assertFalse(res["success"])
+        self.assertEqual(res.get("reason_code"), "preexisting_dirty_worktree")
+
+        # Prove dirty file is preserved byte-for-byte
+        self.assertTrue(os.path.exists(dirty_file))
+        with open(dirty_file, "r") as f:
+            self.assertEqual(f.read(), "dirty edits\n")
+
+    def test_compensating_recovery_on_failed_phase(self):
+        """AC4/Item 4: Failure during transition rolls back ONLY newly created artifacts."""
+        key = "test_key_recovery_1"
+        # Simulate partial progress in journal
+        journal = {
+            "idempotency_key": key,
+            "issue_number": 850,
+            "branch_name": "fix/issue-850-recovery-test",
+            "worktree_path": os.path.join(self.branches_dir, "fix-issue-850-recovery-test"),
+            "artifacts_created": {
+                "branch_created": True,
+                "worktree_dir_created": True,
+                "worktree_registered": True,
+                "lock_created": False,
+            },
+            "failure_reason": "simulated lock failure",
+            "current_phase": author_issue_bootstrap.PHASE_5_REGISTRATION_VERIFIED,
+            "completed": False,
+        }
+        # Create the branch and worktree manually to simulate partial state
+        subprocess.run(["git", "-C", self.repo_dir, "branch", journal["branch_name"]], check=True, capture_output=True)
+        subprocess.run(["git", "-C", self.repo_dir, "worktree", "add", journal["worktree_path"], journal["branch_name"]], check=True, capture_output=True)
+
+        # Run compensating recovery
+        rec = author_issue_bootstrap.run_compensating_recovery(journal, self.repo_dir)
+        self.assertTrue(rec["executed"])
+        self.assertIn(f"worktree_path:{journal['worktree_path']}", rec["rolled_back"])
+        self.assertIn(f"branch:{journal['branch_name']}", rec["rolled_back"])
+
+        # Prove worktree directory and branch were rolled back
+        self.assertFalse(os.path.exists(journal["worktree_path"]))
+        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")
+        self.assertEqual(task_capability_map.required_permission("bootstrap_author_issue_worktree"), "gitea.branch.create")
+        self.assertTrue(task_capability_map.preflight_task_matches("work_issue", "bootstrap_author_issue_worktree"))
+        self.assertTrue(task_capability_map.preflight_task_matches("bootstrap_author_issue_worktree", "lock_issue"))
+
+
+if __name__ == "__main__":
+    unittest.main()

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"

tests/test_branch_cleanup_guard.py

@@ -1266,6 +1266,378 @@ class TestSecondRemediationIntegration(unittest.TestCase):
         self.assertIn("delete_acknowledged", delete_actions[0])
         self.assertTrue(delete_actions[0].get("verified_absent"))
 
+    def test_issue_851_worktree_removed_when_remote_blocked_only_by_worktree_binding(self):
+        """#851: remote blocked by worktree_binding must not skip safe worktree removal.
+
+        Lifecycle: remove clean owned worktree → reassess ownership → delete
+        remote only if independently safe. Unrelated entries stay untouched.
+        """
+        from mcp_server import gitea_reconcile_merged_cleanups
+
+        target_branch = "fix/issue-844-exclude-epic-containers"
+        foreign_branch = "fix/issue-999-unrelated-active"
+        worktree_path = "/tmp/branches/fix-issue-844-exclude-epic-containers"
+        ownership_calls = []
+        remove_calls = []
+        delete_api_calls = []
+
+        def fake_collect(**kwargs):
+            ownership_calls.append(dict(kwargs))
+            # Ownership is reassessed *after* independent worktree removal (#851).
+            # Target worktree is already gone → no worktree_binding remains.
+            # Foreign branch keeps an active author lease → remote delete blocked.
+            if kwargs.get("branch") == foreign_branch:
+                # Match session-bound org/repo + host used by the tool resolve path.
+                return {
+                    "records": [
+                        {
+                            "category": guard.OWNERSHIP_CATEGORY_AUTHOR_LEASE,
+                            "status": "active",
+                            "remote": kwargs.get("remote") or "prgs",
+                            "host": kwargs.get("host") or "gitea.example.com",
+                            "org": kwargs.get("org") or "Scaled-Tech-Consulting",
+                            "repo": kwargs.get("repo") or "Gitea-Tools",
+                            "branch": foreign_branch,
+                            "reclaim_allowed": False,
+                        }
+                    ],
+                    "inventory_error": False,
+                }
+            return {"records": [], "inventory_error": False}
+
+        def fake_remove(project_root, branch, worktree_path=None):
+            remove_calls.append(
+                {"branch": branch, "worktree_path": worktree_path}
+            )
+            return {
+                "success": True,
+                "performed": True,
+                "message": f"removed worktree {worktree_path}",
+                "worktree_path": worktree_path,
+            }
+
+        def fake_probe(h, o, r, auth, br):
+            return guard.classify_branch_readback_http_status(
+                404, not_found_scope=guard.NOT_FOUND_SCOPE_BRANCH
+            )
+
+        def fake_api(method, url, auth, **kwargs):
+            if method == "DELETE":
+                delete_api_calls.append(url)
+            return {}
+
+        report = {
+            "entries": [
+                {
+                    "pr_number": 848,
+                    "head_branch": target_branch,
+                    "remote_branch": {"safe_to_delete_remote": True},
+                    "local_worktree": {
+                        "safe_to_remove_worktree": True,
+                        "worktree_path": worktree_path,
+                    },
+                },
+                {
+                    "pr_number": 999,
+                    "head_branch": foreign_branch,
+                    "remote_branch": {"safe_to_delete_remote": True},
+                    "local_worktree": {
+                        "safe_to_remove_worktree": False,
+                        "worktree_path": None,
+                    },
+                },
+            ],
+            "reviewer_scratch_entries": [],
+        }
+        patch(
+            "mcp_server.get_profile",
+            return_value={
+                "profile_name": "prgs-reconciler",
+                "role": "reconciler",
+                "allowed_operations": [
+                    "gitea.read",
+                    "gitea.branch.delete",
+                    "gitea.pr.close",
+                ],
+                "forbidden_operations": [],
+            },
+        ).start()
+        patch("mcp_server.api_get_all", return_value=[]).start()
+        patch(
+            "mcp_server.merged_cleanup_reconcile.build_reconciliation_report",
+            return_value=report,
+        ).start()
+        patch(
+            "mcp_server.merged_cleanup_reconcile.discover_reviewer_scratch_worktrees",
+            return_value=[],
+        ).start()
+        patch(
+            "mcp_server.audit_reconciliation_mode.check_cleanup_execution_allowed",
+            return_value=(True, []),
+        ).start()
+        patch("mcp_server.verify_preflight_purity", return_value=None).start()
+        patch(
+            "mcp_server._collect_branch_ownership_records",
+            side_effect=fake_collect,
+        ).start()
+        patch("mcp_server._probe_remote_branch", side_effect=fake_probe).start()
+        patch(
+            "mcp_server.merged_cleanup_reconcile.remove_local_worktree",
+            side_effect=fake_remove,
+        ).start()
+        self.mock_api.side_effect = fake_api
+
+        res = gitea_reconcile_merged_cleanups(
+            dry_run=False,
+            execute_confirmed=True,
+            remote="prgs",
+        )
+        self.assertTrue(res.get("performed") or res.get("executed"))
+        actions = res.get("actions") or []
+
+        remove_actions = [
+            a for a in actions if a.get("action") == "remove_local_worktree"
+        ]
+        self.assertEqual(len(remove_actions), 1, actions)
+        self.assertTrue(remove_actions[0].get("success"))
+        self.assertEqual(remove_calls[0]["branch"], target_branch)
+        self.assertEqual(remove_calls[0]["worktree_path"], worktree_path)
+
+        # Target remote delete succeeds after worktree removal + reassessment.
+        target_deletes = [
+            a
+            for a in actions
+            if a.get("action") == "delete_remote_branch"
+            and a.get("branch") == target_branch
+        ]
+        self.assertEqual(len(target_deletes), 1, actions)
+        self.assertTrue(target_deletes[0].get("success"))
+        self.assertTrue(target_deletes[0].get("after_worktree_removal"))
+        self.assertTrue(target_deletes[0].get("ownership_reassessed"))
+        self.assertTrue(target_deletes[0].get("verified_absent"))
+
+        # Foreign branch remains protected (author lease) and is not deleted.
+        foreign_deletes = [
+            a
+            for a in actions
+            if a.get("action") == "delete_remote_branch"
+            and a.get("branch") == foreign_branch
+        ]
+        self.assertEqual(len(foreign_deletes), 1, actions)
+        self.assertFalse(foreign_deletes[0].get("success"))
+        self.assertEqual(
+            foreign_deletes[0].get("blocker_kind"), "active_branch_ownership"
+        )
+        self.assertIn(
+            guard.OWNERSHIP_CATEGORY_AUTHOR_LEASE,
+            foreign_deletes[0].get("blocking_categories") or [],
+        )
+        # Only the target branch should hit the DELETE API.
+        self.assertEqual(len(delete_api_calls), 1)
+
+        # Ownership collected for target (post-removal) and foreign; worktree
+        # removal happened before target remote delete in the action log.
+        target_idx = next(
+            i
+            for i, a in enumerate(actions)
+            if a.get("action") == "remove_local_worktree"
+        )
+        delete_idx = next(
+            i
+            for i, a in enumerate(actions)
+            if a.get("action") == "delete_remote_branch"
+            and a.get("branch") == target_branch
+            and a.get("success")
+        )
+        self.assertLess(target_idx, delete_idx)
+
+    def test_issue_851_dirty_worktree_not_removed_and_remote_stays_protected(self):
+        """#851: dirty/foreign worktrees remain protected; no unsafe cleanup."""
+        from mcp_server import gitea_reconcile_merged_cleanups
+
+        branch = "fix/issue-851-dirty"
+        remove_calls = []
+
+        def fake_collect(**kwargs):
+            return {
+                "records": [
+                    {
+                        "category": guard.OWNERSHIP_CATEGORY_WORKTREE_BINDING,
+                        "status": "active",
+                        "remote": kwargs.get("remote") or "prgs",
+                        "host": kwargs.get("host") or "gitea.example.com",
+                        "org": kwargs.get("org") or "Scaled-Tech-Consulting",
+                        "repo": kwargs.get("repo") or "Gitea-Tools",
+                        "branch": branch,
+                        "reclaim_allowed": False,
+                    }
+                ],
+                "inventory_error": False,
+            }
+
+        report = {
+            "entries": [
+                {
+                    "pr_number": 851,
+                    "head_branch": branch,
+                    "remote_branch": {"safe_to_delete_remote": True},
+                    "local_worktree": {
+                        "safe_to_remove_worktree": False,
+                        "worktree_path": "/tmp/dirty-wt",
+                    },
+                }
+            ],
+            "reviewer_scratch_entries": [],
+        }
+        patch(
+            "mcp_server.get_profile",
+            return_value={
+                "profile_name": "prgs-reconciler",
+                "role": "reconciler",
+                "allowed_operations": [
+                    "gitea.read",
+                    "gitea.branch.delete",
+                ],
+                "forbidden_operations": [],
+            },
+        ).start()
+        patch("mcp_server.api_get_all", return_value=[]).start()
+        patch(
+            "mcp_server.merged_cleanup_reconcile.build_reconciliation_report",
+            return_value=report,
+        ).start()
+        patch(
+            "mcp_server.merged_cleanup_reconcile.discover_reviewer_scratch_worktrees",
+            return_value=[],
+        ).start()
+        patch(
+            "mcp_server.audit_reconciliation_mode.check_cleanup_execution_allowed",
+            return_value=(True, []),
+        ).start()
+        patch("mcp_server.verify_preflight_purity", return_value=None).start()
+        patch(
+            "mcp_server._collect_branch_ownership_records",
+            side_effect=fake_collect,
+        ).start()
+        patch(
+            "mcp_server.merged_cleanup_reconcile.remove_local_worktree",
+            side_effect=lambda *a, **k: remove_calls.append(k) or {
+                "success": True,
+                "performed": True,
+            },
+        ).start()
+        self.mock_api.side_effect = lambda *a, **k: {}
+
+        res = gitea_reconcile_merged_cleanups(
+            dry_run=False,
+            execute_confirmed=True,
+            remote="prgs",
+        )
+        actions = res.get("actions") or []
+        self.assertEqual(remove_calls, [])
+        self.assertFalse(
+            any(a.get("action") == "remove_local_worktree" for a in actions)
+        )
+        deletes = [
+            a for a in actions if a.get("action") == "delete_remote_branch"
+        ]
+        self.assertEqual(len(deletes), 1)
+        self.assertFalse(deletes[0].get("success"))
+        self.assertEqual(deletes[0].get("blocker_kind"), "active_branch_ownership")
+        self.assertIn(
+            guard.OWNERSHIP_CATEGORY_WORKTREE_BINDING,
+            deletes[0].get("blocking_categories") or [],
+        )
+
+    def test_issue_851_idempotent_resume_when_worktree_already_absent(self):
+        """#851: partial failures remain resumable and idempotent."""
+        from mcp_server import gitea_reconcile_merged_cleanups
+
+        branch = "fix/issue-851-resume"
+        ownership_calls = []
+
+        def fake_collect(**kwargs):
+            ownership_calls.append(kwargs)
+            return {"records": [], "inventory_error": False}
+
+        def fake_remove(project_root, branch, worktree_path=None):
+            return {
+                "success": False,
+                "performed": False,
+                "message": f"worktree not found: {worktree_path}",
+            }
+
+        def fake_probe(h, o, r, auth, br):
+            return guard.classify_branch_readback_http_status(
+                404, not_found_scope=guard.NOT_FOUND_SCOPE_BRANCH
+            )
+
+        report = {
+            "entries": [
+                {
+                    "pr_number": 851,
+                    "head_branch": branch,
+                    "remote_branch": {"safe_to_delete_remote": True},
+                    "local_worktree": {
+                        "safe_to_remove_worktree": True,
+                        "worktree_path": "/tmp/already-gone",
+                    },
+                }
+            ],
+            "reviewer_scratch_entries": [],
+        }
+        patch(
+            "mcp_server.get_profile",
+            return_value={
+                "profile_name": "prgs-reconciler",
+                "role": "reconciler",
+                "allowed_operations": [
+                    "gitea.read",
+                    "gitea.branch.delete",
+                ],
+                "forbidden_operations": [],
+            },
+        ).start()
+        patch("mcp_server.api_get_all", return_value=[]).start()
+        patch(
+            "mcp_server.merged_cleanup_reconcile.build_reconciliation_report",
+            return_value=report,
+        ).start()
+        patch(
+            "mcp_server.merged_cleanup_reconcile.discover_reviewer_scratch_worktrees",
+            return_value=[],
+        ).start()
+        patch(
+            "mcp_server.audit_reconciliation_mode.check_cleanup_execution_allowed",
+            return_value=(True, []),
+        ).start()
+        patch("mcp_server.verify_preflight_purity", return_value=None).start()
+        patch(
+            "mcp_server._collect_branch_ownership_records",
+            side_effect=fake_collect,
+        ).start()
+        patch("mcp_server._probe_remote_branch", side_effect=fake_probe).start()
+        patch(
+            "mcp_server.merged_cleanup_reconcile.remove_local_worktree",
+            side_effect=fake_remove,
+        ).start()
+        self.mock_api.side_effect = lambda *a, **k: {}
+
+        res = gitea_reconcile_merged_cleanups(
+            dry_run=False,
+            execute_confirmed=True,
+            remote="prgs",
+        )
+        actions = res.get("actions") or []
+        removes = [a for a in actions if a.get("action") == "remove_local_worktree"]
+        deletes = [a for a in actions if a.get("action") == "delete_remote_branch"]
+        self.assertEqual(len(removes), 1)
+        self.assertFalse(removes[0].get("success"))
+        self.assertEqual(len(deletes), 1)
+        self.assertTrue(deletes[0].get("success"))
+        self.assertTrue(deletes[0].get("after_worktree_removal"))
+        self.assertTrue(ownership_calls)
+
 
 
 if __name__ == "__main__":

tests/test_issue_843_cross_role_handoff.py

@@ -0,0 +1,682 @@
+"""Cross-role allocation handoff consumable by independent workers (#843).
+
+Regression coverage for the controller→required-role consume path:
+
+* controller allocates author work; independent author adopts successfully
+* author adoption succeeds after allocating controller process exits
+* author adoption without sharing controller session identity
+* wrong-role adoption rejected
+* concurrent/second adoption rejected without state corruption
+* terminal allocation adoption rejected
+* successful adoption produces authoritative ownership evidence
+* genuine abandoned-lease recovery remains valid
+* process_work_queue / allocate results include consume identifiers
+* same-role allocation behavior remains compatible
+"""
+
+from __future__ import annotations
+
+import os
+import tempfile
+import unittest
+from datetime import timedelta
+from unittest.mock import patch
+
+from allocator_service import (
+    ALLOCATION_MODE_CROSS_ROLE,
+    ALLOCATION_MODE_ROLE_SCOPED,
+    OUTCOME_ASSIGNED,
+    ROLE_AUTHOR,
+    ROLE_CONTROLLER,
+    ROLE_REVIEWER,
+    WorkCandidate,
+    allocate_next_work,
+)
+from control_plane_db import ControlPlaneDB, ForeignLeaseError, _ts, _utc_now
+import lease_lifecycle as ll
+
+
+class CrossRoleHandoffTest(unittest.TestCase):
+    def setUp(self) -> None:
+        self._tmp = tempfile.TemporaryDirectory()
+        self.db_path = os.path.join(self._tmp.name, "cp.sqlite3")
+        self.db = ControlPlaneDB(self.db_path)
+        self.db.upsert_session(
+            session_id="ctrl-session",
+            role="controller",
+            profile="prgs-controller",
+            pid=99999999,  # dead-looking pid
+        )
+        self.db.upsert_session(
+            session_id="author-worker",
+            role="author",
+            profile="prgs-author",
+            pid=os.getpid(),
+        )
+        self.db.upsert_session(
+            session_id="author-worker-2",
+            role="author",
+            profile="prgs-author",
+            pid=os.getpid(),
+        )
+        self.db.upsert_session(
+            session_id="reviewer-worker",
+            role="reviewer",
+            profile="prgs-reviewer",
+            pid=os.getpid(),
+        )
+        self.wt = self._tmp.name
+
+    def tearDown(self) -> None:
+        self._tmp.cleanup()
+
+    def _ready_issue(self, number: int = 843, title: str = "handoff target") -> WorkCandidate:
+        return WorkCandidate(
+            kind="issue",
+            number=number,
+            labels=("status:ready", "type:bug"),
+            title=title,
+            priority=20,
+        )
+
+    def _controller_allocate(self, number: int = 843, **kwargs):
+        defaults = dict(
+            db=self.db,
+            session_id="ctrl-session",
+            role=ROLE_CONTROLLER,
+            remote="prgs",
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+            candidates=[self._ready_issue(number)],
+            apply=True,
+            profile_name="prgs-controller",
+            username="controller-user",
+            allocation_mode=ALLOCATION_MODE_CROSS_ROLE,
+        )
+        defaults.update(kwargs)
+        return allocate_next_work(**defaults)
+
+    def test_controller_allocates_author_independent_author_adopts(self) -> None:
+        res = self._controller_allocate()
+        self.assertEqual(res["outcome"], OUTCOME_ASSIGNED)
+        self.assertEqual(res["required_role"], ROLE_AUTHOR)
+        self.assertIn("consume_allocation", res)
+        consume = res["consume_allocation"]
+        self.assertEqual(consume["tool"], "gitea_adopt_workflow_lease")
+        self.assertEqual(consume["required_role"], ROLE_AUTHOR)
+        self.assertFalse(consume["controller_session_required"])
+        lid = res["assignment"]["lease_id"]
+        self.assertEqual(consume["lease_id"], lid)
+        self.assertIn(lid, res["next_valid_command"])
+
+        adopted = ll.adopt_lease(
+            self.db,
+            lease_id=lid,
+            adopter_session_id="author-worker",
+            role=ROLE_AUTHOR,
+            worktree_path=self.wt,
+        )
+        self.assertTrue(adopted["success"])
+        self.assertEqual(adopted["outcome"], "adopted_cross_role_handoff")
+        self.assertEqual(adopted["adopted_by_session_id"], "author-worker")
+        self.assertEqual(adopted["adopted_from_session_id"], "ctrl-session")
+        raw = adopted["read_after_write"]
+        self.assertEqual(raw["session_id"], "author-worker")
+        self.assertEqual(raw["adopted_by_session_id"], "author-worker")
+        self.assertEqual(raw["status"], "active")
+        self.assertEqual(raw["phase"], "adopted")
+
+        # Authoritative re-read
+        state = self.db.get_lease_workflow_state(lid)
+        self.assertEqual(state["lease"]["session_id"], "author-worker")
+        self.assertEqual(state["lease"]["adopted_by_session_id"], "author-worker")
+        self.assertEqual(state["assignment"]["session_id"], "author-worker")
+        self.assertEqual(state["provenance"]["handoff_status"], "adopted")
+
+    def test_author_adoption_after_controller_process_exits(self) -> None:
+        res = self._controller_allocate(number=900)
+        lid = res["assignment"]["lease_id"]
+        # Force owner_pid dead + freshness stale_dead_process
+        import sqlite3
+
+        conn = sqlite3.connect(self.db_path)
+        try:
+            conn.execute(
+                "UPDATE leases SET owner_pid = 99999999 WHERE lease_id = ?",
+                (lid,),
+            )
+            conn.commit()
+        finally:
+            conn.close()
+        state = self.db.get_lease_workflow_state(lid)
+        fr = ll.classify_lease_freshness(
+            state["lease"], pid_checker=lambda _p: False
+        )
+        self.assertEqual(fr["freshness"], "stale_dead_process")
+
+        adopted = ll.adopt_lease(
+            self.db,
+            lease_id=lid,
+            adopter_session_id="author-worker",
+            role=ROLE_AUTHOR,
+            worktree_path=self.wt,
+        )
+        self.assertEqual(adopted["outcome"], "adopted_cross_role_handoff")
+        self.assertEqual(adopted["adopted_by_session_id"], "author-worker")
+        # No abandon required
+        state2 = self.db.get_lease_workflow_state(lid)
+        self.assertEqual(state2["lease"]["status"], "active")
+        self.assertNotEqual(state2["lease"]["status"], "abandoned")
+
+    def test_adoption_without_sharing_controller_session_identity(self) -> None:
+        res = self._controller_allocate(number=901)
+        lid = res["assignment"]["lease_id"]
+        adopted = ll.adopt_lease(
+            self.db,
+            lease_id=lid,
+            adopter_session_id="author-worker",
+            role=ROLE_AUTHOR,
+            worktree_path=self.wt,
+        )
+        self.assertNotEqual(adopted["adopted_by_session_id"], "ctrl-session")
+        self.assertFalse(adopted["same_owner"])
+        self.assertEqual(adopted["adopted_from_session_id"], "ctrl-session")
+
+    def test_wrong_role_adoption_rejected(self) -> None:
+        res = self._controller_allocate(number=902)
+        lid = res["assignment"]["lease_id"]
+        with self.assertRaises(ll.LeaseLifecycleError) as ctx:
+            ll.adopt_lease(
+                self.db,
+                lease_id=lid,
+                adopter_session_id="reviewer-worker",
+                role=ROLE_REVIEWER,
+            )
+        self.assertIn("wrong role", str(ctx.exception).lower())
+        # State unchanged
+        state = self.db.get_lease_workflow_state(lid)
+        self.assertEqual(state["lease"]["session_id"], "ctrl-session")
+        self.assertIsNone(state["lease"].get("adopted_by_session_id") or None)
+        self.assertEqual(state["provenance"]["handoff_status"], "pending")
+
+    def test_second_adoption_rejected_without_corruption(self) -> None:
+        res = self._controller_allocate(number=903)
+        lid = res["assignment"]["lease_id"]
+        first = ll.adopt_lease(
+            self.db,
+            lease_id=lid,
+            adopter_session_id="author-worker",
+            role=ROLE_AUTHOR,
+            worktree_path=self.wt,
+        )
+        self.assertEqual(first["outcome"], "adopted_cross_role_handoff")
+        with self.assertRaises(ll.LeaseLifecycleError):
+            ll.adopt_lease(
+                self.db,
+                lease_id=lid,
+                adopter_session_id="author-worker-2",
+                role=ROLE_AUTHOR,
+                worktree_path=self.wt,
+            )
+        state = self.db.get_lease_workflow_state(lid)
+        self.assertEqual(state["lease"]["session_id"], "author-worker")
+        self.assertEqual(state["lease"]["adopted_by_session_id"], "author-worker")
+        self.assertEqual(state["assignment"]["session_id"], "author-worker")
+        self.assertEqual(state["lease"]["status"], "active")
+
+    def test_terminal_allocation_adoption_rejected(self) -> None:
+        res = self._controller_allocate(number=904)
+        lid = res["assignment"]["lease_id"]
+        # Abandon as terminal
+        proof = ll.AbandonProof(
+            dead_process=True,
+            missing_worktree=True,
+            no_open_pr=True,
+            no_live_mutation_risk=True,
+            owner_pid=99999999,
+            worktree_path="/nonexistent/for-843",
+        )
+        # Attach dead pid / missing wt for abandon eligibility
+        import sqlite3
+
+        conn = sqlite3.connect(self.db_path)
+        try:
+            conn.execute(
+                "UPDATE leases SET owner_pid = 99999999, worktree_path = ? WHERE lease_id = ?",
+                ("/nonexistent/for-843", lid),
+            )
+            conn.commit()
+        finally:
+            conn.close()
+        abandoned = ll.abandon_lease(
+            self.db,
+            lease_id=lid,
+            requester_session_id="author-worker",
+            proof=proof,
+        )
+        self.assertEqual(abandoned["outcome"], "abandoned")
+        with self.assertRaises(ll.LeaseLifecycleError) as ctx:
+            ll.adopt_lease(
+                self.db,
+                lease_id=lid,
+                adopter_session_id="author-worker",
+                role=ROLE_AUTHOR,
+            )
+        self.assertIn("abandoned", str(ctx.exception).lower())
+
+    def test_successful_adoption_read_after_write_ownership(self) -> None:
+        res = self._controller_allocate(number=905)
+        lid = res["assignment"]["lease_id"]
+        adopted = ll.adopt_lease(
+            self.db,
+            lease_id=lid,
+            adopter_session_id="author-worker",
+            role=ROLE_AUTHOR,
+            worktree_path=self.wt,
+        )
+        raw = adopted["read_after_write"]
+        self.assertEqual(raw["lease_id"], lid)
+        self.assertEqual(raw["session_id"], "author-worker")
+        self.assertEqual(raw["adopted_by_session_id"], "author-worker")
+        self.assertEqual(raw["adopted_from_session_id"], "ctrl-session")
+        # Re-fetch proves durable write
+        state = self.db.get_lease_workflow_state(lid)
+        self.assertEqual(state["lease"]["session_id"], raw["session_id"])
+        self.assertEqual(
+            state["lease"]["adopted_by_session_id"], raw["adopted_by_session_id"]
+        )
+
+    def test_genuine_abandoned_recovery_still_valid(self) -> None:
+        """Same-role author lease abandoned remains reclaimable via abandon path."""
+        same = allocate_next_work(
+            self.db,
+            session_id="author-worker",
+            role=ROLE_AUTHOR,
+            remote="prgs",
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+            candidates=[self._ready_issue(906, "same-role")],
+            apply=True,
+            profile_name="prgs-author",
+            username="author-user",
+            allocation_mode=ALLOCATION_MODE_ROLE_SCOPED,
+        )
+        self.assertEqual(same["outcome"], OUTCOME_ASSIGNED)
+        lid = same["assignment"]["lease_id"]
+        import sqlite3
+
+        conn = sqlite3.connect(self.db_path)
+        try:
+            conn.execute(
+                "UPDATE leases SET owner_pid = 99999999, worktree_path = ? WHERE lease_id = ?",
+                ("/nonexistent/same-role", lid),
+            )
+            conn.commit()
+        finally:
+            conn.close()
+        proof = ll.AbandonProof(
+            dead_process=True,
+            missing_worktree=True,
+            no_open_pr=True,
+            no_live_mutation_risk=True,
+            owner_pid=99999999,
+            worktree_path="/nonexistent/same-role",
+        )
+        abandoned = ll.abandon_lease(
+            self.db,
+            lease_id=lid,
+            requester_session_id="author-worker-2",
+            proof=proof,
+        )
+        self.assertEqual(abandoned["outcome"], "abandoned")
+        # Foreign author cannot handoff-consume an abandoned non-handoff lease
+        with self.assertRaises(ll.LeaseLifecycleError):
+            ll.adopt_lease(
+                self.db,
+                lease_id=lid,
+                adopter_session_id="author-worker-2",
+                role=ROLE_AUTHOR,
+            )
+        # Reclaim path still works for expired/abandoned after force-expire
+        reclaimed = ll.reclaim_expired_lease(
+            self.db,
+            lease_id=lid,
+            session_id="author-worker-2",
+            role=ROLE_AUTHOR,
+            worktree_path=self.wt,
+        )
+        self.assertEqual(reclaimed["outcome"], "reclaimed")
+        self.assertEqual(reclaimed["assignment"]["session_id"], "author-worker-2")
+
+    def test_allocate_payload_includes_consume_identifiers(self) -> None:
+        res = self._controller_allocate(number=907)
+        self.assertIn("consume_allocation", res)
+        c = res["consume_allocation"]
+        for key in (
+            "tool",
+            "lease_id",
+            "assignment_id",
+            "required_role",
+            "required_profile",
+            "required_namespace",
+            "instructions",
+            "handoff_status",
+        ):
+            self.assertIn(key, c)
+        self.assertEqual(c["required_namespace"], "gitea-author")
+        self.assertEqual(c["required_profile"], "prgs-author")
+        self.assertIn("gitea_adopt_workflow_lease", c["instructions"])
+        self.assertTrue(res["lease_proof"]["cross_role_handoff"])
+        self.assertEqual(res["lease_proof"]["handoff_status"], "pending")
+
+    def test_same_role_allocation_remains_compatible(self) -> None:
+        res = allocate_next_work(
+            self.db,
+            session_id="author-worker",
+            role=ROLE_AUTHOR,
+            remote="prgs",
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+            candidates=[self._ready_issue(908)],
+            apply=True,
+            profile_name="prgs-author",
+            username="author-user",
+        )
+        self.assertEqual(res["outcome"], OUTCOME_ASSIGNED)
+        self.assertNotIn("consume_allocation", res)
+        lid = res["assignment"]["lease_id"]
+        state = self.db.get_lease_workflow_state(lid)
+        # No cross-role handoff provenance
+        prov = state.get("provenance") or {}
+        self.assertFalse(prov.get("cross_role_handoff"))
+        # Owner resume still works
+        resume = ll.adopt_lease(
+            self.db,
+            lease_id=lid,
+            adopter_session_id="author-worker",
+            role=ROLE_AUTHOR,
+            worktree_path=self.wt,
+        )
+        self.assertTrue(resume["same_owner"])
+        self.assertEqual(resume["outcome"], "adopted_owner_resume")
+
+    def test_inspect_points_required_role_at_consume(self) -> None:
+        res = self._controller_allocate(number=909)
+        lid = res["assignment"]["lease_id"]
+        decision = ll.inspect_lease(
+            self.db, lid, caller_session_id="author-worker"
+        )
+        self.assertEqual(
+            decision["safe_next_action"], ll.SAFE_CONSUME_CROSS_ROLE
+        )
+        self.assertFalse(decision["block"])
+        self.assertEqual(decision["required_role"], ROLE_AUTHOR)
+
+    def test_db_cas_rejects_concurrent_second_consume(self) -> None:
+        res = self._controller_allocate(number=910)
+        lid = res["assignment"]["lease_id"]
+        # First consume via DB layer directly
+        first = self.db.adopt_lease(
+            lease_id=lid,
+            adopter_session_id="author-worker",
+            role=ROLE_AUTHOR,
+            worktree_path=self.wt,
+            provenance={
+                "cross_role_handoff": True,
+                "handoff_status": "adopted",
+                "required_role": "author",
+            },
+        )
+        self.assertEqual(first["outcome"], "adopted_cross_role_handoff")
+        # Second CAS must fail
+        with self.assertRaises(ForeignLeaseError):
+            self.db.adopt_lease(
+                lease_id=lid,
+                adopter_session_id="author-worker-2",
+                role=ROLE_AUTHOR,
+                worktree_path=self.wt,
+                provenance={
+                    "cross_role_handoff": True,
+                    "handoff_status": "pending",
+                    "required_role": "author",
+                },
+            )
+        state = self.db.get_lease_workflow_state(lid)
+        self.assertEqual(state["lease"]["session_id"], "author-worker")
+
+
+class MCPBoundaryAdoptRoleBindingTest(unittest.TestCase):
+    """#843 F1: MCP-boundary role binding for ``gitea_adopt_workflow_lease``.
+
+    The library-level wrong-role test calls ``lease_lifecycle.adopt_lease``
+    directly. These tests prove the MCP entry point derives the adopter role
+    authoritatively from the active authenticated profile and rejects any
+    caller-supplied role that disagrees, so a reviewer/merger profile cannot
+    consume an author handoff by passing ``role="author"``.
+    """
+
+    AUTHOR_PROFILE = {
+        "profile_name": "prgs-author",
+        "role": "author",
+        "allowed_operations": [
+            "gitea.read",
+            "gitea.pr.create",
+            "gitea.branch.push",
+        ],
+        "forbidden_operations": [],
+    }
+    REVIEWER_PROFILE = {
+        "profile_name": "prgs-reviewer",
+        "role": "reviewer",
+        "allowed_operations": [
+            "gitea.read",
+            "gitea.pr.review",
+            "gitea.pr.approve",
+            "gitea.pr.request_changes",
+        ],
+        "forbidden_operations": ["gitea.pr.create", "gitea.branch.push"],
+    }
+    MERGER_PROFILE = {
+        "profile_name": "prgs-merger",
+        "role": "merger",
+        "allowed_operations": ["gitea.read", "gitea.pr.merge"],
+        "forbidden_operations": ["gitea.pr.create", "gitea.branch.push"],
+    }
+    FOREIGN_AUTHOR_PROFILE = {
+        "profile_name": "dadeschools-author",
+        "role": "author",
+        "allowed_operations": [
+            "gitea.read",
+            "gitea.pr.create",
+            "gitea.branch.push",
+        ],
+        "forbidden_operations": [],
+    }
+
+    def setUp(self) -> None:
+        self._tmp = tempfile.TemporaryDirectory()
+        self.db_path = os.path.join(self._tmp.name, "cp.sqlite3")
+        self.db = ControlPlaneDB(self.db_path)
+        self.db.upsert_session(
+            session_id="ctrl-session",
+            role="controller",
+            profile="prgs-controller",
+            pid=99999999,
+        )
+        self.db.upsert_session(
+            session_id="author-worker",
+            role="author",
+            profile="prgs-author",
+            pid=os.getpid(),
+        )
+        self.wt = self._tmp.name
+
+    def tearDown(self) -> None:
+        self._tmp.cleanup()
+
+    def _ready_issue(self, number: int) -> WorkCandidate:
+        return WorkCandidate(
+            kind="issue",
+            number=number,
+            labels=("status:ready", "type:bug"),
+            title="handoff target",
+            priority=20,
+        )
+
+    def _handoff_lease(self, number: int = 843) -> str:
+        res = allocate_next_work(
+            db=self.db,
+            session_id="ctrl-session",
+            role=ROLE_CONTROLLER,
+            remote="prgs",
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+            candidates=[self._ready_issue(number)],
+            apply=True,
+            profile_name="prgs-controller",
+            username="controller-user",
+            allocation_mode=ALLOCATION_MODE_CROSS_ROLE,
+        )
+        self.assertEqual(res["outcome"], OUTCOME_ASSIGNED)
+        self.assertEqual(res["required_role"], ROLE_AUTHOR)
+        return res["assignment"]["lease_id"]
+
+    def _call_adopt_tool(self, profile: dict, **kwargs):
+        import gitea_mcp_server as mcp_server
+
+        with (
+            patch.object(mcp_server, "get_profile", return_value=profile),
+            patch.object(
+                mcp_server,
+                "_control_plane_db_or_error",
+                return_value=(self.db, []),
+            ),
+        ):
+            return mcp_server.gitea_adopt_workflow_lease(
+                remote="prgs", **kwargs
+            )
+
+    def _assert_handoff_untouched(self, lease_id: str) -> None:
+        state = self.db.get_lease_workflow_state(lease_id)
+        self.assertEqual(state["lease"]["session_id"], "ctrl-session")
+        self.assertIsNone(state["lease"].get("adopted_by_session_id") or None)
+        self.assertEqual(state["lease"]["status"], "active")
+        self.assertEqual(state["provenance"]["handoff_status"], "pending")
+
+    def test_reviewer_profile_cannot_consume_author_handoff_via_role_author(
+        self,
+    ) -> None:
+        lid = self._handoff_lease(920)
+        result = self._call_adopt_tool(
+            self.REVIEWER_PROFILE,
+            lease_id=lid,
+            session_id="reviewer-worker",
+            role="author",
+            worktree_path=self.wt,
+        )
+        self.assertFalse(result["success"])
+        self.assertEqual(result["outcome"], "blocked")
+        self.assertEqual(result["profile_role_kind"], "reviewer")
+        self.assertEqual(result["supplied_role"], "author")
+        self.assertIn("does not match", result["reasons"][0])
+        self._assert_handoff_untouched(lid)
+
+    def test_merger_profile_cannot_consume_author_handoff_via_role_author(
+        self,
+    ) -> None:
+        lid = self._handoff_lease(921)
+        result = self._call_adopt_tool(
+            self.MERGER_PROFILE,
+            lease_id=lid,
+            session_id="merger-worker",
+            role="author",
+            worktree_path=self.wt,
+        )
+        self.assertFalse(result["success"])
+        self.assertEqual(result["outcome"], "blocked")
+        self.assertEqual(result["profile_role_kind"], "merger")
+        self._assert_handoff_untouched(lid)
+
+    def test_reviewer_profile_rejected_without_role_argument(self) -> None:
+        """Even without a spoofed role, the profile-derived role binds."""
+        lid = self._handoff_lease(922)
+        result = self._call_adopt_tool(
+            self.REVIEWER_PROFILE,
+            lease_id=lid,
+            session_id="reviewer-worker",
+            worktree_path=self.wt,
+        )
+        self.assertFalse(result["success"])
+        self.assertEqual(result["outcome"], "blocked")
+        self.assertIn("wrong role", result["reasons"][0].lower())
+        self._assert_handoff_untouched(lid)
+
+    def test_author_profile_mismatching_supplied_role_rejected(self) -> None:
+        lid = self._handoff_lease(923)
+        result = self._call_adopt_tool(
+            self.AUTHOR_PROFILE,
+            lease_id=lid,
+            session_id="author-worker",
+            role="reviewer",
+            worktree_path=self.wt,
+        )
+        self.assertFalse(result["success"])
+        self.assertEqual(result["outcome"], "blocked")
+        self.assertEqual(result["profile_role_kind"], "author")
+        self.assertEqual(result["supplied_role"], "reviewer")
+        self._assert_handoff_untouched(lid)
+
+    def test_foreign_profile_name_rejected_for_author_handoff(self) -> None:
+        """Provenance required_profile binds even when the role matches."""
+        lid = self._handoff_lease(924)
+        result = self._call_adopt_tool(
+            self.FOREIGN_AUTHOR_PROFILE,
+            lease_id=lid,
+            session_id="foreign-author-worker",
+            worktree_path=self.wt,
+        )
+        self.assertFalse(result["success"])
+        self.assertEqual(result["outcome"], "blocked")
+        self.assertIn("wrong profile", result["reasons"][0].lower())
+        self._assert_handoff_untouched(lid)
+
+    def test_author_profile_consumes_author_handoff(self) -> None:
+        lid = self._handoff_lease(925)
+        result = self._call_adopt_tool(
+            self.AUTHOR_PROFILE,
+            lease_id=lid,
+            session_id="author-worker",
+            role="author",
+            worktree_path=self.wt,
+        )
+        self.assertTrue(result["success"])
+        self.assertEqual(result["outcome"], "adopted_cross_role_handoff")
+        self.assertEqual(result["adopted_by_session_id"], "author-worker")
+        self.assertEqual(result["adopted_from_session_id"], "ctrl-session")
+        state = self.db.get_lease_workflow_state(lid)
+        self.assertEqual(state["lease"]["session_id"], "author-worker")
+        self.assertEqual(
+            state["lease"]["adopted_by_session_id"], "author-worker"
+        )
+        self.assertEqual(state["assignment"]["session_id"], "author-worker")
+        self.assertEqual(state["provenance"]["handoff_status"], "adopted")
+
+    def test_author_profile_consumes_author_handoff_without_role_argument(
+        self,
+    ) -> None:
+        lid = self._handoff_lease(926)
+        result = self._call_adopt_tool(
+            self.AUTHOR_PROFILE,
+            lease_id=lid,
+            session_id="author-worker",
+            worktree_path=self.wt,
+        )
+        self.assertTrue(result["success"])
+        self.assertEqual(result["outcome"], "adopted_cross_role_handoff")
+        state = self.db.get_lease_workflow_state(lid)
+        self.assertEqual(state["lease"]["session_id"], "author-worker")
+        self.assertEqual(state["lease"]["role"], "author")
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_merged_cleanup_reconcile.py

@@ -12,6 +12,59 @@ import merged_cleanup_reconcile as mcr  # noqa: E402
 
 
 class TestMergedCleanupAssessment(unittest.TestCase):
+    def test_issue_851_plan_order_worktree_then_reassess_then_remote(self):
+        """#851 dry-run plan: remove worktree, reassess ownership, then remote."""
+        plan = mcr.plan_cleanup_execution_order(
+            remote_assessment={"safe_to_delete_remote": True},
+            local_assessment={"safe_to_remove_worktree": True},
+        )
+        actions = [s["action"] for s in plan]
+        self.assertEqual(
+            actions,
+            [
+                "remove_local_worktree",
+                "reassess_branch_ownership",
+                "delete_remote_branch",
+            ],
+        )
+        self.assertEqual(plan[0]["phase"], 1)
+        self.assertEqual(plan[-1]["phase"], 3)
+        self.assertIn("independently_safe", plan[0]["reason"])
+        self.assertIn("reassessment", plan[-1]["reason"])
+
+    def test_issue_851_plan_remote_only_when_worktree_not_safe(self):
+        plan = mcr.plan_cleanup_execution_order(
+            remote_assessment={"safe_to_delete_remote": True},
+            local_assessment={"safe_to_remove_worktree": False},
+        )
+        self.assertEqual([s["action"] for s in plan], ["delete_remote_branch"])
+        self.assertNotIn("reassess_branch_ownership", [s["action"] for s in plan])
+
+    def test_issue_851_plan_worktree_only_when_remote_not_safe(self):
+        plan = mcr.plan_cleanup_execution_order(
+            remote_assessment={"safe_to_delete_remote": False},
+            local_assessment={"safe_to_remove_worktree": True},
+        )
+        self.assertEqual([s["action"] for s in plan], ["remove_local_worktree"])
+
+    def test_issue_851_entry_includes_planned_execution_order(self):
+        entry = mcr.build_pr_cleanup_entry(
+            pr={
+                "number": 848,
+                "title": "Closes #844",
+                "body": "",
+                "merged_at": "2026-07-23T00:00:00Z",
+                "head": {"ref": "fix/issue-844-x", "sha": "a" * 40},
+            },
+            project_root="/tmp/not-a-real-root",
+            open_pr_heads=set(),
+            remote_branch_exists=True,
+            head_on_master=True,
+            delete_capability_allowed=True,
+        )
+        self.assertIn("planned_execution_order", entry)
+        self.assertIsInstance(entry["planned_execution_order"], list)
+
     def test_extract_linked_issue_from_closes(self):
         issue = mcr.extract_linked_issue(
             "feat: cleanup (Closes #269)",

tests/test_task_capability_role_invariants.py

@@ -139,6 +139,8 @@ EXPECTED_ROLE_EXCLUSIVE_TASKS = frozenset(
         "gitea_release_merger_pr_lease",
         "create_branch",
         "push_branch",
+        "bootstrap_author_issue_worktree",
+        "gitea_bootstrap_author_issue_worktree",
         # #812 AC20: publishing an unpublished local head is author-only for the
         # same reason every other push is — it writes a branch to the remote.
         "publish_unpublished_branch",

tests/test_webui_shell.py

@@ -0,0 +1,135 @@
+"""Tests for the Phase 1 operator console application shell (#638)."""
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from starlette.routing import Route
+from starlette.testclient import TestClient
+
+from webui import layout
+from webui.app import create_app
+from webui.nav import NAV_GROUPS, STUB_PAGES, nav_hrefs
+
+
+class TestShellNav(unittest.TestCase):
+    def setUp(self):
+        self.client = TestClient(create_app())
+
+    def test_nav_group_labels_present(self):
+        text = self.client.get("/").text
+        for group in NAV_GROUPS:
+            with self.subTest(group=group.label):
+                self.assertIn(f">{group.label}<", text)
+
+    def test_phase1_group_labels_cover_expected_ia(self):
+        labels = {group.label for group in NAV_GROUPS}
+        for expected in (
+            "Health",
+            "Traffic",
+            "Runtime/Sessions",
+            "Projects",
+            "Inventory",
+            "Timeline",
+            "Policy",
+            "Insights",
+        ):
+            with self.subTest(label=expected):
+                self.assertIn(expected, labels)
+
+    def test_every_nav_href_resolves_to_a_get_route(self):
+        app = create_app()
+        get_paths = {
+            route.path
+            for route in app.routes
+            if isinstance(route, Route) and "GET" in route.methods
+        }
+        for href in nav_hrefs():
+            with self.subTest(href=href):
+                self.assertIn(href, get_paths, f"nav href {href} has no GET route")
+
+    def test_legacy_hrefs_still_navigable(self):
+        text = self.client.get("/").text
+        for href in ("/queue", "/projects", "/prompts", "/runtime",
+                     "/audit", "/worktrees", "/leases", "/actions"):
+            with self.subTest(href=href):
+                self.assertIn(f'href="{href}"', text)
+
+
+class TestShellBadges(unittest.TestCase):
+    def setUp(self):
+        self.client = TestClient(create_app())
+
+    def test_mode_badge_present(self):
+        self.assertIn("mode: read-only", self.client.get("/").text)
+
+    def test_environment_badge_present(self):
+        self.assertIn("env:", self.client.get("/").text)
+
+    def test_default_environment_is_local(self):
+        self.assertEqual(layout.environment_label(), "local")
+
+    def test_remote_bind_reports_remote_environment(self):
+        import os
+
+        prior = os.environ.get("WEBUI_HOST")
+        os.environ["WEBUI_HOST"] = "10.0.0.5"
+        try:
+            self.assertEqual(layout.environment_label(), "remote")
+        finally:
+            if prior is None:
+                os.environ.pop("WEBUI_HOST", None)
+            else:
+                os.environ["WEBUI_HOST"] = prior
+
+    def test_docs_link_present(self):
+        text = self.client.get("/").text
+        self.assertIn(layout.DOCS_URL, text)
+        self.assertIn(">Docs<", text)
+
+
+class TestShellStubs(unittest.TestCase):
+    def setUp(self):
+        self.client = TestClient(create_app())
+
+    def test_stub_routes_render_200(self):
+        for path, (title, _desc) in STUB_PAGES.items():
+            with self.subTest(path=path):
+                response = self.client.get(path)
+                self.assertEqual(response.status_code, 200, path)
+                self.assertIn(title, response.text)
+                self.assertIn("placeholder", response.text)
+
+    def test_stub_routes_are_read_only(self):
+        for path in STUB_PAGES:
+            with self.subTest(path=path):
+                response = self.client.post(path)
+                self.assertEqual(response.status_code, 405)
+                self.assertEqual(response.json()["error"], "read-only-mvp")
+
+    def test_stub_pages_carry_nav_and_badges(self):
+        response = self.client.get("/inventory")
+        self.assertIn("mode: read-only", response.text)
+        self.assertIn('href="/queue"', response.text)
+
+
+class TestShellHome(unittest.TestCase):
+    def setUp(self):
+        self.client = TestClient(create_app())
+
+    def test_home_summarizes_console(self):
+        text = self.client.get("/").text
+        self.assertIn("Operator console", text)
+        self.assertIn("Phase 1", text)
+
+    def test_home_links_legacy_pages(self):
+        text = self.client.get("/").text
+        self.assertIn("MVP legacy pages", text)
+        for href in ("/queue", "/audit", "/leases"):
+            with self.subTest(href=href):
+                self.assertIn(f'href="{href}"', text)
+
+
+if __name__ == "__main__":
+    unittest.main()

webui/app.py

@@ -12,6 +12,7 @@ from starlette.routing import Route
 
 from webui.deployment_boundary import deployment_snapshot
 from webui.layout import render_page
+from webui.nav import NAV_GROUPS, STUB_PAGES
 from webui.project_registry import (
     ProjectRegistry,
     RegistryError,
@@ -65,24 +66,62 @@ def _stub_page(title: str, description: str) -> HTMLResponse:
     return HTMLResponse(render_page(title=title, body_html=body))
 
 
+_LEGACY_PAGES = (
+    ("/queue", "Queue", "live PR and issue dashboard (#429)"),
+    ("/projects", "Projects", "registry and onboarding (#427)"),
+    ("/prompts", "Prompts", "canonical workflow prompt library (#428)"),
+    ("/runtime", "Runtime", "MCP health and stale-runtime detection (#430)"),
+    ("/audit", "Audit", "final-report paste and validator preview (#431)"),
+    ("/worktrees", "Worktrees", "branch hygiene dashboard (#432)"),
+    ("/leases", "Leases", "collision and lease visibility (#433)"),
+    ("/actions", "Actions", "gated write-action framework (#434)"),
+)
+
+
+def _render_home_nav_groups() -> str:
+    groups = []
+    for group in NAV_GROUPS:
+        items = "".join(
+            f'<li><a href="{item.href}">{item.label}</a>'
+            + ("" if item.status == "live" else " <span class=\"muted\">(stub)</span>")
+            + "</li>"
+            for item in group.items
+        )
+        groups.append(f"<h3>{group.label}</h3><ul>{items}</ul>")
+    return "".join(groups)
+
+
 async def home(_request: Request) -> HTMLResponse:
+    legacy = "".join(
+        f"<li><strong>{label}</strong> — {desc} "
+        f'(<a href="{href}">{href}</a>)</li>'
+        for href, label, desc in _LEGACY_PAGES
+    )
     body = (
         "<h2>Operator console</h2>"
-        "<p>Local entry point for MCP Control Plane operational views.</p>"
+        "<p>Read-only home for the MCP Control Plane Phase 1 operator console. "
-        "<ul>"
+        "Gitea, MCP capability gates, and canonical workflows remain the source "
-        "<li><strong>Queue</strong> — live PR and issue dashboard (#429)</li>"
+        "of truth; this console never mutates them.</p>"
-        "<li><strong>Projects</strong> — registry and onboarding (#427)</li>"
+        "<h2>Phase 1 surfaces</h2>"
-        "<li><strong>Prompts</strong> — canonical workflow prompt library (#428)</li>"
+        + _render_home_nav_groups()
-        "<li><strong>Runtime</strong> — MCP health and stale-runtime detection (#430)</li>"
+        + "<h2>MVP legacy pages</h2>"
-        "<li><strong>Audit</strong> — final-report paste and validator preview (#431)</li>"
+        "<ul>" + legacy + "</ul>"
-        "<li><strong>Worktrees</strong> — branch hygiene dashboard (#432)</li>"
-        "<li><strong>Leases</strong> — collision and lease visibility (#433)</li>"
-        "<li><strong>Actions</strong> — gated write-action framework (#434)</li>"
-        "</ul>"
     )
     return HTMLResponse(render_page(title="Home", body_html=body))
 
 
+async def phase_stub(request: Request) -> HTMLResponse:
+    """Graceful read-only placeholder for a not-yet-implemented Phase 1 surface."""
+    title, description = STUB_PAGES[request.url.path]
+    body = (
+        f"<h2>{title}</h2>"
+        f'<div class="stub"><p>{description}</p>'
+        "<p>Phase 1 shell placeholder — no write actions. Tracked under "
+        "epic #631.</p></div>"
+    )
+    return HTMLResponse(render_page(title=title, body_html=body))
+
+
 async def health(_request: Request) -> JSONResponse:
     """Liveness only — deliberately cheap, runs no dependency probe (#634).
 
@@ -472,6 +511,10 @@ def create_app(*, bind_host: str | None = None) -> Starlette:
                 api_console_security_model,
                 methods=["GET"],
             ),
+            *[
+                Route(path, phase_stub, methods=["GET"])
+                for path in STUB_PAGES
+            ],
         ],
         exception_handlers={405: method_not_allowed},
     )

webui/layout.py

@@ -2,28 +2,66 @@
 
 from __future__ import annotations
 
-NAV_ITEMS = (
+import os
-    ("/", "Home"),
+
-    ("/queue", "Queue"),
+from webui.nav import NAV_GROUPS
-    ("/projects", "Projects"),
-    ("/prompts", "Prompts"),
-    ("/runtime", "Runtime"),
-    ("/audit", "Audit"),
-    ("/worktrees", "Worktrees"),
-    ("/leases", "Leases"),
-    ("/actions", "Actions"),
-)
 
 MVP_NOTICE = (
     "Read-only MVP — Gitea, MCP tools, and canonical workflows remain the "
     "source of truth. No mutation endpoints."
 )
 
+# Canonical docs entry point surfaced from the shell header (#638).
+DOCS_URL = (
+    "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/src/branch/"
+    "master/docs/webui-local-dev.md"
+)
+
+_LOCAL_HOSTS = frozenset({"", "127.0.0.1", "localhost", "::1"})
+
+
+def environment_label() -> str:
+    """Classify the serving environment as ``local`` or ``remote`` (#638).
+
+    Derived from the same ``WEBUI_HOST`` default the app binds to; loopback
+    hosts are ``local``, anything else is ``remote``. Read-only signal only.
+    """
+    host = (os.environ.get("WEBUI_HOST", "127.0.0.1") or "").strip().lower()
+    return "local" if host in _LOCAL_HOSTS else "remote"
+
+
+def _render_nav() -> str:
+    groups_html = []
+    for group in NAV_GROUPS:
+        links = "".join(
+            f'<a href="{item.href}"'
+            + (' class="nav-stub"' if item.status == "stub" else "")
+            + f'>{item.label}</a>'
+            for item in group.items
+        )
+        groups_html.append(
+            '<div class="nav-group">'
+            f'<span class="nav-group-label">{group.label}</span>'
+            f'<span class="nav-group-links">{links}</span>'
+            "</div>"
+        )
+    return "".join(groups_html)
+
+
+def _render_badges() -> str:
+    env = environment_label()
+    return (
+        '<div class="header-badges">'
+        f'<span class="badge env-badge env-{env}">env: {env}</span>'
+        '<span class="badge mode-badge">mode: read-only</span>'
+        f'<a class="badge docs-link" href="{DOCS_URL}">Docs</a>'
+        "</div>"
+    )
+
 
 def render_page(*, title: str, body_html: str, extra_head: str = "") -> str:
-    nav_links = "".join(
+    nav_links = _render_nav()
-        f'<a href="{href}">{label}</a>' for href, label in NAV_ITEMS
+    header_badges = _render_badges()
-    )
     return f"""<!DOCTYPE html>
 <html lang="en">
 <head>
@@ -53,21 +91,58 @@ def render_page(*, title: str, body_html: str, extra_head: str = "") -> str:
       padding: 0.75rem 1.25rem;
     }}
     header h1 {{
-      margin: 0 0 0.5rem;
+      margin: 0;
       font-size: 1.1rem;
       font-weight: 600;
     }}
+    .header-top {{
+      display: flex;
+      flex-wrap: wrap;
+      align-items: center;
+      justify-content: space-between;
+      gap: 0.5rem 1rem;
+      margin-bottom: 0.6rem;
+    }}
+    .header-badges {{ display: inline-flex; flex-wrap: wrap; gap: 0.4rem; }}
+    .env-badge.env-local {{ color: #8fd19e; border-color: #3d6b4a; }}
+    .env-badge.env-remote {{ color: #e0c27a; border-color: #6b5730; }}
+    .mode-badge {{ color: #9ec8f0; border-color: #3d5f7a; }}
+    a.docs-link {{
+      color: var(--accent);
+      border-color: var(--accent);
+      text-decoration: none;
+      text-transform: none;
+    }}
+    a.docs-link:hover {{ filter: brightness(1.12); }}
     nav {{
       display: flex;
       flex-wrap: wrap;
-      gap: 0.75rem 1rem;
+      gap: 0.5rem 1.25rem;
     }}
+    .nav-group {{
+      display: flex;
+      flex-direction: column;
+      gap: 0.15rem;
+    }}
+    .nav-group-label {{
+      font-size: 0.68rem;
+      text-transform: uppercase;
+      letter-spacing: 0.04em;
+      color: var(--muted);
+    }}
+    .nav-group-links {{ display: inline-flex; flex-wrap: wrap; gap: 0.6rem; }}
     nav a {{
       color: var(--accent);
       text-decoration: none;
       font-size: 0.9rem;
     }}
     nav a:hover {{ text-decoration: underline; }}
+    nav a.nav-stub {{ color: var(--muted); }}
+    nav a.nav-stub::after {{
+      content: " ·stub";
+      font-size: 0.7rem;
+      color: var(--muted);
+    }}
     main {{
       max-width: 52rem;
       margin: 0 auto;
@@ -166,7 +241,10 @@ def render_page(*, title: str, body_html: str, extra_head: str = "") -> str:
 </head>
 <body>
   <header>
+    <div class="header-top">
       <h1>MCP Control Plane</h1>
+      {header_badges}
+    </div>
     <nav>{nav_links}</nav>
   </header>
   <main>

webui/nav.py

@@ -0,0 +1,111 @@
+"""Navigation IA for the Phase 1 operator console shell (#638).
+
+Single source of truth for the console navigation so ``webui/layout.py`` and
+the ``webui/app.py`` route table stay aligned with epic #631. Read-only: every
+destination is a GET view or a Phase 1 placeholder. No mutation links.
+
+Nav groups follow the #631 Phase 1 information architecture: Health, Traffic,
+Runtime/Sessions, Projects, Inventory, Timeline, Policy (placeholder), and
+Insights (placeholder). Later-phase surfaces are declared as ``stub`` items and
+backed by ``STUB_PAGES`` so their nav links resolve to a graceful placeholder
+instead of a 404.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True)
+class NavItem:
+    """A single navigation destination.
+
+    ``status`` is ``"live"`` for implemented views and ``"stub"`` for Phase 1
+    placeholders whose backing view lands in a later child issue.
+    """
+
+    href: str
+    label: str
+    status: str = "live"
+
+
+@dataclass(frozen=True)
+class NavGroup:
+    label: str
+    items: tuple[NavItem, ...]
+
+
+NAV_GROUPS: tuple[NavGroup, ...] = (
+    NavGroup("Health", (
+        NavItem("/health", "Liveness"),
+    )),
+    NavGroup("Traffic", (
+        NavItem("/queue", "Queue"),
+        NavItem("/leases", "Leases"),
+        NavItem("/actions", "Actions"),
+    )),
+    NavGroup("Runtime/Sessions", (
+        NavItem("/runtime", "Runtime health"),
+        NavItem("/sessions", "Sessions", "stub"),
+    )),
+    NavGroup("Projects", (
+        NavItem("/projects", "Projects"),
+    )),
+    NavGroup("Inventory", (
+        NavItem("/inventory", "Inventory", "stub"),
+        NavItem("/worktrees", "Worktrees"),
+    )),
+    NavGroup("Timeline", (
+        NavItem("/timeline", "Timeline", "stub"),
+    )),
+    NavGroup("Policy", (
+        NavItem("/policy", "Policy", "stub"),
+        NavItem("/prompts", "Prompts"),
+    )),
+    NavGroup("Insights", (
+        NavItem("/insights", "Insights", "stub"),
+        NavItem("/audit", "Audit"),
+    )),
+)
+
+
+# Phase 1 placeholder destinations whose backing views land in later child
+# issues of epic #631. Each maps a path to (title, description). Routes are
+# registered so nav links resolve to a graceful, read-only stub page.
+STUB_PAGES: dict[str, tuple[str, str]] = {
+    "/sessions": (
+        "Sessions",
+        "Active session, capability, and role inventory. Backed by the unified "
+        "inventory API (#636) once it lands.",
+    ),
+    "/inventory": (
+        "Inventory",
+        "Unified sessions, leases, locks, namespaces, and worktree inventory. "
+        "Backed by the Phase 1 inventory API (#636).",
+    ),
+    "/timeline": (
+        "Timeline",
+        "Workflow event timeline across issues and PRs. A later Phase 1 surface.",
+    ),
+    "/policy": (
+        "Policy",
+        "Capability and role policy surface. Placeholder until a later phase.",
+    ),
+    "/insights": (
+        "Insights",
+        "Aggregate operational insights and trends. Placeholder until a later "
+        "phase.",
+    ),
+}
+
+
+def iter_nav_items():
+    """Yield every ``NavItem`` across all groups in declared order."""
+    for group in NAV_GROUPS:
+        for item in group.items:
+            yield item
+
+
+def nav_hrefs() -> tuple[str, ...]:
+    """Return every navigation href in declared order."""
+    return tuple(item.href for item in iter_nav_items())