Merge branch 'master' into fix/issue-855-pr-scoped-merged-cleanup

Merge pull request 'fix(author): refresh durable issue-lock head after branch sync; recover merge-sync-drifted dead-session locks (Closes #871 )' (#874 ) from fix/issue-871-durable-lock-head-refresh into master
Merge branch 'master' into fix/issue-871-durable-lock-head-refresh
2026-07-24 03:15:43 -05:00 · 2026-07-24 03:04:24 -05:00 · 2026-07-24 02:42:52 -05:00 · 2026-07-24 02:26:02 -05:00 · 2026-07-24 02:07:33 -05:00 · 2026-07-24 02:07:17 -05:00
21 changed files with 2751 additions and 1850 deletions
@@ -1,976 +0,0 @@
 """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,
    journal_dir: str | None = None,
 ) -> dict[str, Any]:
    """Execute compensating recovery for artifacts created by this transition only."""
    artifacts = journal.get("artifacts_created") or {}
    pending = journal.get("pending_creations") or {}
    rolled_back: list[str] = []
    worktree_path = journal.get("worktree_path")
    branch_name = journal.get("branch_name")
    # Roll back issue lock if created
    if artifacts.get("lock_created") or pending.get("lock"):
        issue_num = journal.get("issue_number")
        session_id = journal.get("owner_session")
        if issue_num and session_id:
            try:
                issue_lock_store.release_session_lock(
                    issue_number=issue_num,
                    session=session_id,
                    lock_dir=journal_dir,
                )
                rolled_back.append(f"lock:issue-{issue_num}")
            except Exception:
                pass
        artifacts["lock_created"] = False
    worktree_created = (
        artifacts.get("worktree_registered")
        or artifacts.get("worktree_dir_created")
        or (pending.get("worktree_path") == worktree_path and worktree_path)
    )
    if worktree_created and worktree_path:
        if os.path.exists(worktree_path):
            # Re-verify cleanliness before destructive removal (F-4)
            porc_res = subprocess.run(
                ["git", "-C", worktree_path, "status", "--porcelain"],
                capture_output=True,
                text=True,
                check=False,
            )
            is_dirty = porc_res.returncode == 0 and bool(porc_res.stdout.strip())
            if is_dirty:
                rolled_back.append(f"worktree_path_preserved_dirty:{worktree_path}")
            else:
                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}")
        else:
            rolled_back.append(f"worktree_path:{worktree_path}")
    branch_created = (
        artifacts.get("branch_created")
        or (pending.get("branch_name") == branch_name and branch_name)
    )
    if 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:
                # Check for author commits on branch before branch deletion (F-4)
                resolved_base = journal.get("resolved_base_sha") or "master"
                rev_list_res = subprocess.run(
                    [
                        "git",
                        "-C",
                        canonical_repo_root,
                        "rev-list",
                        f"{resolved_base}..{branch_name}",
                    ],
                    capture_output=True,
                    text=True,
                    check=False,
                )
                has_commits = rev_list_res.returncode == 0 and bool(rev_list_res.stdout.strip())
                if has_commits:
                    rolled_back.append(f"branch_preserved_commits:{branch_name}")
                else:
                    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, journal_dir=journal_dir)
    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
 import stat
 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 = os.path.realpath(get_journal_dir(journal_dir))
        if not os.path.isdir(lock_dir):
            raise RuntimeError(f"Lock directory '{lock_dir}' does not exist or is not a directory")
        raw_lock_path = os.path.abspath(os.path.join(lock_dir, f"{safe_key}.lock"))
        try:
            common = os.path.commonpath([lock_dir, os.path.dirname(raw_lock_path)])
        except Exception:
            common = None
        if common != lock_dir:
            raise RuntimeError(f"Lock path '{raw_lock_path}' escapes canonical lock directory '{lock_dir}'")
        self.lock_path = raw_lock_path
        self.fd = None
    def __enter__(self):
        if os.path.islink(self.lock_path):
            raise RuntimeError(f"Refusing lock acquisition: lock path '{self.lock_path}' is a symlink")
        flags = os.O_RDWR | os.O_CREAT
        if hasattr(os, "O_NOFOLLOW"):
            flags |= os.O_NOFOLLOW
        if hasattr(os, "O_CLOEXEC"):
            flags |= os.O_CLOEXEC
        try:
            fd = os.open(self.lock_path, flags, 0o600)
        except OSError as exc:
            raise RuntimeError(f"Failed to open lock file safely '{self.lock_path}': {exc}") from exc
        st = os.fstat(fd)
        if not stat.S_ISREG(st.st_mode):
            os.close(fd)
            raise RuntimeError(f"Lock target '{self.lock_path}' is not a regular file")
        self.fd = fd
        fcntl.flock(self.fd, fcntl.LOCK_EX)
        return self
    def __exit__(self, exc_type, exc_val, exc_tb):
        if self.fd is not None:
            try:
                fcntl.flock(self.fd, fcntl.LOCK_UN)
            except Exception:
                pass
            try:
                os.close(self.fd)
            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)
    session = (owner_session or "").strip()
    if not session:
        return {
            "success": False,
            "reason_code": "missing_owner_session",
            "message": "Missing required owner_session parameter (fail closed). Session identifier cannot be fabricated or defaulted.",
            "exact_next_action": (
                "Pass explicit owner_session resolved from gitea_whoami or session context."
            ),
        }
    if active_identity is None or not str(active_identity).strip():
        return {
            "success": False,
            "reason_code": "missing_active_identity",
            "message": "Missing required active_identity parameter (fail closed). Identity cannot be fabricated or defaulted.",
            "exact_next_action": "Pass explicit active_identity resolved from gitea_whoami.",
        }
    identity = str(active_identity).strip()
    if active_profile is None or not str(active_profile).strip():
        return {
            "success": False,
            "reason_code": "missing_active_profile",
            "message": "Missing required active_profile parameter (fail closed). Profile cannot be fabricated or defaulted.",
            "exact_next_action": "Pass explicit active_profile resolved from gitea_whoami.",
        }
    profile = str(active_profile).strip()
    # 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, journal_dir=lock_dir):
        # Idempotency check
        existing = load_phase_journal(key, journal_dir=lock_dir)
        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": (
                        "Call gitea_whoami, then gitea_resolve_task_capability(task='work_issue') "
                        "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": identity,
                "active_profile": profile,
                "owner_session": 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, journal_dir=lock_dir)
                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, journal_dir=lock_dir)
        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)
        pending_branch = (journal.get("pending_creations") or {}).get("branch_name")
        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():
                # F-8: Check if branch shares a common ancestor with live master
                mb_check = subprocess.run(
                    ["git", "-C", root, "merge-base", live_master_sha, branch_head],
                    capture_output=True,
                    text=True,
                    check=False,
                )
                if mb_check.returncode != 0 or not mb_check.stdout.strip():
                    journal["failure_reason"] = (
                        f"existing branch '{target_branch}' HEAD ({branch_head[:12]}) is incompatible with live master ({live_master_sha[:12]})"
                    )
                    save_phase_journal(journal, journal_dir=lock_dir)
                    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 sync the existing branch with master before bootstrapping."
                        ),
                    }
            # Preserve creation provenance monotonically across interruption and replay
            if was_branch_created_previously or pending_branch == target_branch:
                journal["artifacts_created"]["branch_created"] = True
            else:
                journal["artifacts_created"]["branch_created"] = False
        else:
            # Persist creation intent/provenance to disk BEFORE executing external mutation
            journal.setdefault("pending_creations", {})["branch_name"] = target_branch
            journal["artifacts_created"]["branch_created"] = True
            save_phase_journal(journal, journal_dir=lock_dir)
            # 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["artifacts_created"]["branch_created"] = False
                journal.get("pending_creations", {}).pop("branch_name", None)
                journal["failure_reason"] = (
                    f"failed to create git branch '{target_branch}': {create_res.stderr.strip()}"
                )
                save_phase_journal(journal, journal_dir=lock_dir)
                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["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, journal_dir=lock_dir)
        # 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, journal_dir=lock_dir)
            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)
        pending_wt = (journal.get("pending_creations") or {}).get("worktree_path")
        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, journal_dir=lock_dir)
                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, journal_dir=lock_dir)
                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."
                    ),
                }
            if was_dir_created_previously or pending_wt == target_worktree:
                journal["artifacts_created"]["worktree_dir_created"] = True
                journal["artifacts_created"]["worktree_registered"] = True
            else:
                journal["artifacts_created"]["worktree_dir_created"] = False
                journal["artifacts_created"]["worktree_registered"] = False
        else:
            # Persist creation intent/provenance to disk BEFORE executing external worktree add mutation
            journal.setdefault("pending_creations", {})["worktree_path"] = target_worktree
            journal["artifacts_created"]["worktree_dir_created"] = True
            journal["artifacts_created"]["worktree_registered"] = True
            save_phase_journal(journal, journal_dir=lock_dir)
            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["artifacts_created"]["worktree_dir_created"] = False
                journal["artifacts_created"]["worktree_registered"] = False
                journal.get("pending_creations", {}).pop("worktree_path", None)
                journal["failure_reason"] = (
                    f"git worktree add failed: {wt_add_res.stderr.strip()}"
                )
                run_compensating_recovery(journal, root, journal_dir=lock_dir)
                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["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, journal_dir=lock_dir)
        # 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, journal_dir=lock_dir)
            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, journal_dir=lock_dir)
        # 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": session,
                "claimant": {
                    "username": identity,
                    "profile": profile,
                },
                "assignment_id": assignment_id,
                "lease_id": lease_id,
                "expected_base_sha": live_master_sha,
                "created_at": datetime.now(timezone.utc).isoformat(),
            }
            journal.setdefault("pending_creations", {})["lock"] = True
            journal["artifacts_created"]["lock_created"] = True
            save_phase_journal(journal, journal_dir=lock_dir)
            lock_res = issue_lock_store.bind_session_lock(lock_data, lock_dir=lock_dir)
        except Exception as exc:
            journal["artifacts_created"]["lock_created"] = False
            journal.get("pending_creations", {}).pop("lock", None)
            journal["failure_reason"] = f"issue lock binding failed: {exc}"
            run_compensating_recovery(journal, root, journal_dir=lock_dir)
            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, journal_dir=lock_dir)
        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": (
                "Call gitea_whoami, then gitea_resolve_task_capability(task='work_issue') "
                "and proceed with author implementation in the bootstrapped worktree."
            ),
        }
@@ -40,66 +40,22 @@ def _normalize_path(path: str) -> str:
    return (path or "").replace("\\", "/").rstrip("/")
 def get_canonical_branches_root(project_root: str | None = None) -> str:
    """Return the absolute path of the canonical branches directory for *project_root*."""
    root = os.path.realpath(project_root) if project_root else os.path.realpath(os.getcwd())
    canonical_repo_root = resolve_canonical_repo_root(root, root)
    return os.path.realpath(os.path.join(canonical_repo_root, "branches"))
 def is_path_under_branches(path: str, project_root: str | None = None) -> bool:
-    """True when *path* resolves inside a canonical ``branches/`` directory."""
+    """True when *path* resolves inside ``<project_root>/branches/``."""
-    if not path or not str(path).strip():
+    normalized = _normalize_path(path)
    if not normalized:
        return False
-    try:
+    if "/branches/" in f"{normalized}/":
-        real_path = os.path.realpath(os.path.abspath(str(path).strip()))
+        return True
-    except Exception:
+    if normalized.endswith("/branches"):
-        return False
+        return True
-
+    if project_root:
-    branches_root = get_canonical_branches_root(project_root or real_path)
+        root = _normalize_path(os.path.realpath(project_root))
-    try:
+        real = _normalize_path(os.path.realpath(path))
-        common = os.path.commonpath([branches_root, real_path])
+        if real.startswith(f"{root}/"):
-    except Exception:
+            rel = real[len(root) + 1 :]
-        return False
+            return rel == "branches" or rel.startswith("branches/")
-
+    return False
    if common != branches_root:
        return False
    rel = os.path.relpath(real_path, branches_root)
    return rel != "." and not rel.startswith("..")
 def resolve_canonical_repo_root(workspace_path: str, fallback_project_root: str) -> str:
    """Return the stable repository root for *workspace_path* via git metadata (#460)."""
    p = (workspace_path or "").strip()
    if p:
        try:
            res = subprocess.run(
                ["git", "-C", p, "rev-parse", "--git-common-dir"],
                capture_output=True,
                text=True,
                check=True,
            )
            common = _realpath_git_common_dir(p, res.stdout)
            if common.endswith(f"{os.sep}.git") or os.path.basename(common) == ".git":
                candidate_root = os.path.dirname(common)
                real_p = os.path.realpath(p)
                try:
                    if os.path.commonpath([candidate_root, real_p]) == candidate_root:
                        return candidate_root
                except Exception:
                    pass
        except Exception:
            pass
    fallback = os.path.realpath(fallback_project_root or workspace_path or ".")
    norm = fallback.replace("\\", "/")
    if "/branches/" in norm:
        return os.path.realpath(norm.split("/branches/")[0])
    elif norm.endswith("/branches"):
        return os.path.realpath(os.path.dirname(fallback))
    return fallback
 def resolve_mutation_workspace(
@@ -131,6 +87,29 @@ def _realpath_git_common_dir(workspace_path: str, common_dir: str) -> str:
    return os.path.realpath(os.path.join(workspace_path, raw))
 def resolve_canonical_repo_root(workspace_path: str, fallback_project_root: str) -> str:
    """Return the stable repository root for *workspace_path* via git metadata (#460)."""
    path = (workspace_path or "").strip()
    fallback = os.path.realpath(fallback_project_root)
    if not path:
        return fallback
    try:
        res = subprocess.run(
            ["git", "-C", path, "rev-parse", "--git-common-dir"],
            capture_output=True,
            text=True,
            check=True,
        )
        common = _realpath_git_common_dir(path, res.stdout)
    except Exception:
        return fallback
    if common.endswith(f"{os.sep}.git"):
        return os.path.dirname(common)
    if os.path.basename(common) == ".git":
        return os.path.dirname(common)
    return fallback
 def resolve_author_mutation_context(
    worktree_path: str | None,
    process_project_root: str,
@@ -525,6 +525,90 @@ def assess_ownership_record_activity(record: dict[str, Any]) -> dict[str, Any]:
    }
 # Reviewer-lease reclaim is only reachable from a non-live (expired/stale) lease.
 _RECLAIMABLE_REVIEWER_STATUSES = _EXPIRED_STATUSES | _STALE_STATUSES
 def is_active_ownership_status(status: str | None) -> bool:
    """True when *status* denotes live/active ownership of a branch (#855).
    Used to decide whether a *competing* active claimant still uses a branch
    when weighing an expired reviewer lease for reclaim. Expired, stale,
    released, and terminal statuses are not active.
    """
    return _norm_str(status).lower() in _ACTIVE_OWNERSHIP_STATUSES
 def assess_expired_reviewer_lease_reclaim(
    *,
    role: str,
    status: str,
    pr_merged: bool | None,
    owner_pid_alive: bool | None,
    competing_active_claimant: bool | None,
 ) -> dict[str, Any]:
    """Decide, explicitly and fail-closed, whether an expired reviewer lease
    may stop protecting an already-merged branch (#855 AC4).
    An expired reviewer lease should not protect a merged branch forever once
    its work is done and no live claimant remains. Reclaim is permitted only
    when **every** condition below is provably satisfied; any unknown
    (``None``) or contrary value keeps the lease protective:
    - the lease is a ``reviewer`` lease (author/merger/controller/reconciler
      leases are out of scope and always keep protecting);
    - its status is expired or stale (never an active/live lease);
    - the PR is proven merged (``pr_merged is True``);
    - the lease owner process is proven dead (``owner_pid_alive is False``);
    - no competing active claimant uses the branch
      (``competing_active_claimant is False``).
    Returns a decision dict with ``reclaim_allowed`` and, when refused, the
    fail-closed ``reasons``. The reasons never contain secrets — only the
    role, the status, and which condition was unproven.
    """
    reasons: list[str] = []
    normalized_role = _norm_str(role).lower()
    normalized_status = _norm_str(status).lower()
    if normalized_role != "reviewer":
        reasons.append(
            f"lease role '{normalized_role or 'unknown'}' is not a reviewer "
            "lease; expired-reviewer reclaim does not apply"
        )
    if normalized_status not in _RECLAIMABLE_REVIEWER_STATUSES:
        reasons.append(
            f"lease status '{normalized_status or 'unknown'}' is not expired "
            "or stale; only a non-live reviewer lease may be reclaimed"
        )
    if pr_merged is not True:
        reasons.append(
            "PR merged state is not proven true; reclaim requires an "
            "already-merged PR (fail closed)"
        )
    if owner_pid_alive is not False:
        reasons.append(
            "lease owner process liveness is not proven dead; a live owner "
            "still protects the branch (fail closed)"
        )
    if competing_active_claimant is not False:
        reasons.append(
            "a competing active claimant may still use the branch; reclaim "
            "requires no other active ownership (fail closed)"
        )
    allowed = not reasons
    return {
        "reclaim_allowed": allowed,
        "role": normalized_role,
        "status": normalized_status,
        "decision": (
            "reclaim_expired_reviewer_lease" if allowed else "keep_protecting"
        ),
        "reasons": [] if allowed else reasons,
    }
 def assess_active_branch_ownership(
    *,
    remote: str,
@@ -247,8 +247,7 @@ def bootstrap_permits_control_checkout(
    """
    if not isinstance(assessment, dict):
        return False
-    import author_issue_bootstrap
+    if not is_create_issue_task(task):
    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
@@ -0,0 +1,223 @@
 # ADR: MCP restart governance and authorization policy
 - **Status:** Accepted (policy effective immediately for LLM and operator sessions; enforcement tooling may lag)
 - **Date:** 2026-07-23
 - **Tracking issue:** [#656](https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/issues/656)
 - **Policy version:** `restart-governance/v1`
 - **Related:**
  - Umbrella: [#655](https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/issues/655) — governed MCP restart coordination and zero-disruption recovery
  - Vision: [#652](https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/issues/652) — MCP Control Plane Web Console product vision (§A system health and process control)
  - Roadmap: [#653](https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/issues/653) — Control Plane Web Console phased delivery (Phase 2 restart controls)
  - Contamination guard: [#630](https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/issues/630) — blocks manual process-kill recovery
  - Console restart UX: [#642](https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/issues/642) — sanctioned restart and graceful reload
  - Existing restart / reconnect paths to inventory: [#591](https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/issues/591) — auto-restart on master advance (closed); [#584](https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/issues/584) — host auto-reconnect on transport flap
  - Stable-control runtime split: `docs/architecture/mcp-stable-control-runtime-policy-adr.md` (#615)
  - Client-namespace health: `docs/mcp-namespace-health.md` (#543)
  - Reconnect-only EOF recovery: `docs/mcp-namespace-eof-recovery.md`
 ## 1. Context
 The Gitea MCP server is the **control plane** for real issue and PR mutations
 (create, comment, lock, review, merge, reconcile). The same process serves every
 role namespace (`gitea-author`, `gitea-reviewer`, `gitea-merger`,
 `gitea-reconciler`, `gitea-controller`) and holds the in-memory capability-gate
 code loaded at startup.
 Restarting that process is destructive to concurrent work:
 - It resets every session's identity, preflight, and capability-lease binding.
 - It can interrupt a mutation mid-critical-section (a lock acquire, a review
  submit, a merge), leaving durable state half-written.
 - Relaunching from the wrong checkout or worktree silently changes which code
  the control plane runs, defeating master-parity gates (#420 / #615).
 Today there is **no durable written policy** stating who may restart MCP, under
 what conditions, that restart is a last resort, and how controller approval,
 automated safety gates, and break-glass interact. Operators and LLM sessions
 therefore invent restart behavior ad hoc, which makes concurrent multi-role work
 unsafe. #630 and #642 need this policy as their backbone.
 This ADR defines that policy. It does **not** implement coordinator code or HA
 multi-instance restart (those are later children of #655).
 ## 2. Decision
 ### 2.1 v1 decision (recorded)
 **Restart authority in v1 is `controller approval + automated safety gates`.**
 A restart of the stable control runtime is authorized only when **both** hold:
 1. A **controller** role explicitly approves the restart, recording an audit
   entry (who, why, scope, affected sessions), **and**
 2. The **automated safety gates** pass: a completed drain acknowledgement (no
   affected session is mid-critical-section) or a declared break-glass incident
   (§2.5).
 Quorum among multiple controllers is **not** required day-one. It is deferred
 unless a later investigation (tracked under #653) proves single-controller
 approval is insufficient. This ADR records the v1 decision so enforcement code
 (#630) has a fixed target; changing it requires a superseding ADR.
 ### 2.2 Restart is a last resort — the recovery ladder
 Restart is the **last** rung. Before any restart, exhaust the narrower
 recoveries, in order:
 1. **Reconnect** the IDE/client MCP namespace (transport EOF, `client is
   closing: EOF`, transient `#584` flap). No process change. See
   `docs/mcp-namespace-eof-recovery.md`.
 2. **Refresh / rebind** the session workspace: re-run `gitea_whoami`,
   `gitea_resolve_task_capability`, and pass an explicit validated
   `worktree_path`. Fixes stale session context without touching the process.
 3. **Scoped restart** of a single misbehaving namespace/service (where the
   deployment supports per-service restart) rather than the whole control plane.
 4. **Full restart** of the stable control runtime process — operator-owned,
   controller-approved, drained.
 5. **Host / infrastructure restart** — the broadest action; same authorization
   as a full restart plus infrastructure ownership.
 A session **must** try rungs 1–2 and record why they were insufficient before
 requesting a restart at rung 3 or above. Skipping straight to restart is a
 policy violation.
 ### 2.3 Authorization matrix
 | Role | Reconnect (1) | Refresh/rebind (2) | Scoped restart (3) | Full restart (4) | Host restart (5) |
 |---|---|---|---|---|---|
 | **author** | self | self | request only | **forbidden** | forbidden |
 | **reviewer** | self | self | request only | **forbidden** | forbidden |
 | **merger** | self | self | request only | **forbidden** | forbidden |
 | **reconciler** | self | self | request only | **forbidden** | forbidden |
 | **controller** | self | self | **approve** (+gates) | **approve** (+gates) | request to operator |
 | **operator** | self | self | execute (controller-approved) | execute (controller-approved) | execute (controller-approved) |
 | **admin** | self | self | execute | execute | execute (break-glass) |
 Legend: *self* = may perform for its own client session; *request only* = may
 raise a restart request but not authorize or execute it; *approve* = may
 authorize under §2.1 gates; *execute* = may perform the process action after the
 authorization is recorded.
 Key invariants:
 - **No LLM worker role (author/reviewer/merger/reconciler) may perform or
  authorize a full or host restart.** They may only reconnect/rebind their own
  client and file a restart request.
 - **Controller approval authorizes; operator/admin executes.** The approving
  controller and the executing operator may be the same human, but both the
  approval and the execution are audited.
 - Privileged process actions (full restart, host restart) are reserved to
  **operator/admin**, never to an automated worker.
 ### 2.4 Approved conditions
 A restart at rung 3+ is approved only under one of these recorded conditions:
 - **No affected sessions:** the control plane has no live session that would be
  interrupted (verified, not assumed).
 - **Full drain acknowledged:** every affected session has drained
  (no open critical section — no held mutation lease mid-write) and the drain is
  acknowledged in the audit record.
 - **Controller + gates:** controller approval plus passing automated safety
  gates (§2.1), the standard v1 path.
 - **Quorum:** not required in v1; reserved for a future superseding ADR.
 - **Break-glass:** an incident-backed emergency exception (§2.5).
 Restart **never** bypasses mutation gates mid-critical-section. Drain before
 restart is mandatory except under break-glass with a declared incident.
 ### 2.5 Break-glass
 Break-glass is a **separate, narrower** authorization path for emergencies where
 the normal drain-and-approve path cannot complete (e.g. the control plane is
 wedged and cannot drain).
 Break-glass conditions:
 - A declared incident record exists (id, timestamp, declarer) **before** the
  action.
 - The action is taken by **operator or admin** authority only — never by an LLM
  worker role, and never unilaterally by an operator with active peers when a
  controller is reachable.
 - The scope is the minimum necessary rung of the ladder.
 - A **mandatory post-hoc audit** entry is filed: what was restarted, why the
  normal path was impossible, which sessions were affected, and the incident id.
 Break-glass suspends the drain requirement, not the audit requirement.
 ### 2.6 Explicit prohibitions
 - **A unilateral LLM or operator full restart while active peer sessions
  exist is forbidden.** An LLM worker role must not kill, restart, or relaunch
  the MCP process; a lone operator must not full-restart over live peer work
  without controller approval or a break-glass incident.
 - Process-kill recovery is forbidden as a routine tool (#630). This ADR does not
  introduce a kill path.
 - Ambiguous policy state **denies** restart (§4).
 ## 3. Security requirements
 - Full restart and host restart are **privileged**; only operator/admin execute
  them, only after a controller approval or break-glass incident is recorded.
 - Break-glass is a distinct authorization path with its own audit mandate; it is
  never the default and never silent.
 - **Every approval and every restart action is audited** (who approved, who
  executed, scope, affected sessions, condition, policy version). No restart is
  authorized without a durable audit entry.
 ## 4. Failure behavior
 **Ambiguous policy → deny restart.** If it cannot be established that a
 restart is authorized under §2 — unknown affected-session state, missing
 controller approval, absent break-glass incident, or an unclassifiable request —
 the safe action is to **refuse** the restart and stop with a recovery report,
 never to restart on assumption.
 ## 5. Policy IDs (for enforcement code)
 Enforcement code — the restart coordinator (a later child of #655), the #630
 contamination guard, and the #642 console restart UX — binds to these stable
 policy identifiers rather than to prose:
 | Policy ID | Statement |
 |---|---|
 | `RG-01` | Restart is last resort; rungs 1–2 must be tried and recorded first (§2.2). |
 | `RG-02` | v1 authority = controller approval + automated safety gates (§2.1). |
 | `RG-03` | No LLM worker role performs or authorizes full/host restart (§2.3). |
 | `RG-04` | Full/host restart executed by operator/admin only, post approval (§2.3). |
 | `RG-05` | Drain before restart is mandatory except break-glass with incident (§2.4). |
 | `RG-06` | Break-glass requires a pre-declared incident and post-hoc audit (§2.5). |
 | `RG-07` | Unilateral LLM/operator full restart with active peers is forbidden (§2.6). |
 | `RG-08` | Ambiguous policy state denies restart (§4). |
 The `restart-governance/v1` **policy version** field is emitted on future
 restart audit events so approvals can be reconciled against the policy revision
 in force.
 ## 6. Dogfooding
 Gitea-Tools governs its own MCP control plane by this policy. Author, reviewer,
 merger, and reconciler sessions operating on this repository use the recovery
 ladder (§2.2) — reconnect and rebind, never self-restart — and any real restart
 of the Gitea-Tools stable control runtime follows the controller-approval +
 drain path defined here.
 ## 7. Acceptance and cross-links
 This ADR is the authoritative restart-governance policy. It **must** stay
 cross-linked from the safety model and the web-console deployment boundary:
 - `docs/safety-model.md` § Process restart governance references this ADR.
 - `docs/webui-deployment.md` references this ADR for restart/reload disposition.
 It is linked to its issue lineage — umbrella **#655**, vision **#652**, roadmap
 **#653**, contamination guard **#630**, and console restart UX **#642** — in
 § Related above.
 ## 8. Non-goals
 - Implementing the restart coordinator or approval state machine (#630, later
  children of #655).
 - Implementing HA multi-instance restart or quorum machinery.
 - Introducing any process-kill or auto-restart tool; existing auto-restart
  behavior must be inventoried before any new restart tool is enabled.
@@ -69,7 +69,6 @@ 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`
@@ -46,3 +46,17 @@ If shell helpers are unavailable and MCP commit cannot run, stop with a recovery
 report (restart session, clear hung terminals, use MCP-native commit). See
 [`llm-workflow-runbooks.md`](llm-workflow-runbooks.md) § MCP-native commit path
 (#260) and agent temp artifact cleanup (#261).
 ## 7. Process restart governance
 Restarting the MCP control-plane process is destructive to concurrent multi-role
 work and is governed by a dedicated policy. Restart is a **last resort** behind
 narrower recoveries (reconnect, rebind), full/host restart is reserved to
 operator/admin under **controller approval + automated safety gates**, a
 unilateral LLM or operator full restart with active peers is **forbidden**, and
 ambiguous policy state **denies** restart. Break-glass is a separate,
 incident-backed path with a mandatory audit.
 See [`architecture/mcp-restart-governance.md`](architecture/mcp-restart-governance.md)
 (#656) for the authorization matrix, the recovery ladder, break-glass
 conditions, and the `RG-01`–`RG-08` policy IDs.
@@ -55,6 +55,15 @@ shipped to the browser.
 assumption paths, and the client-secret policy. Use it to verify an instance is
 configured for internal-only operation.
 ## Process restart / reload disposition
 The console never exposes a restart or reload control; process restart of the
 MCP control-plane runtime is governed separately. Restart is a last resort behind
 reconnect/rebind, full restart is operator/admin-only under controller approval
 plus safety gates, and break-glass is an incident-backed path. See
 [`architecture/mcp-restart-governance.md`](architecture/mcp-restart-governance.md)
 (#656).
 ## Non-goals (MVP)
 - Full SSO or session login in the UI
@@ -977,32 +977,6 @@ 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
@@ -1495,7 +1469,7 @@ def verify_preflight_purity(
            dirty_files = sorted(
                _parse_porcelain_entries(_get_workspace_porcelain(workspace))
            )
-            if dirty_files and task != "commit_files":
+            if dirty_files:
                raise RuntimeError(
                    nwb.format_namespace_workspace_binding_error(
                        role_kind=role,
@@ -2471,6 +2445,20 @@ def _evaluate_issue_lock_recovery(
            descendant_sha=local_head,
        )
    # #871: the inverse of the #768 descendant relation — the *remote* head may
    # have advanced past the local/recorded head via a sanctioned merge-based
    # branch sync (``gitea_update_pr_branch_by_merge``) while the local worktree
    # stayed put. Observe that provenance server-side so the assessor can prove
    # it and nothing else. Probed only when the heads differ; never from any
    # caller-supplied value.
    sync_provenance: dict | None = None
    if remote_head and local_head and remote_head != local_head:
        sync_provenance = issue_lock_worktree.read_merge_sync_provenance(
            worktree_path,
            prior_head_sha=local_head,
            synced_head_sha=remote_head,
        )
    # #772: with no remote branch there is no head to measure against, so the
    # base the branch was cut from is observed instead. Probed only in that
    # case, so the published path's evidence is untouched (#772 AC8).
@@ -2513,6 +2501,7 @@ def _evaluate_issue_lock_recovery(
        remote_branch_exists=remote_branch_exists,
        recorded_base_sha=recorded_base,
        base_ancestry=base_ancestry,
        sync_provenance=sync_provenance,
    )
@@ -9313,7 +9302,6 @@ def gitea_commit_files(
    host: str | None = None,
    org: str | None = None,
    repo: str | None = None,
    worktree_path: str | None = None,
 ) -> dict:
    """Commit changes to multiple files in a Gitea repository in a single atomic commit.
@@ -9326,46 +9314,10 @@ def gitea_commit_files(
        host: Override the Gitea host.
        org: Override the owner/organization.
        repo: Override the repository name.
        worktree_path: Optional worktree path for author mutation context.
    Returns:
        dict with success status and commit/branch information.
    """
    if worktree_path is None:
        lock_data = issue_lock_store.read_session_issue_lock() or {}
        worktree_path = lock_data.get("worktree_path")
        if not worktree_path:
            try:
                prof = get_profile()
                uname = prof.get("username") or prof.get("profile_name")
                for path in issue_lock_store.iter_lock_files():
                    lk = issue_lock_store.read_lock_file(path) or {}
                    claimant = lk.get("claimant") or {}
                    if lk.get("remote") == remote and (claimant.get("username") == uname or lk.get("profile") == prof.get("profile_name")):
                        issue_lock_store.bind_session_lock(lk, renewal_sanctioned=True)
                        worktree_path = lk.get("worktree_path")
                        break
            except Exception:
                pass
    if worktree_path is None and files:
        for f in files:
            p = f.get("workspace_path") or f.get("local_path") or ""
            if p and os.path.isabs(p):
                real_p = os.path.realpath(p)
                real_root = os.path.realpath(PROJECT_ROOT)
                branches_dir = os.path.join(real_root, "branches")
                if real_p.startswith(branches_dir + os.sep):
                    rel_sub = os.path.relpath(real_p, branches_dir)
                    wt_folder = rel_sub.split(os.sep)[0]
                    if wt_folder and wt_folder != "..":
                        worktree_path = os.path.join(branches_dir, wt_folder)
                        break
    if worktree_path:
        os.environ["GITEA_AUTHOR_WORKTREE"] = worktree_path
        os.environ["GITEA_ACTIVE_WORKTREE"] = worktree_path
    ok, block_reasons = role_session_router.check_author_mutation_after_reviewer_stop(
        "commit_files"
    )
@@ -9378,7 +9330,7 @@ def gitea_commit_files(
            "reasons": block_reasons,
        }
    blocked = _namespace_mutation_block(
-        "commit_files", commit="", branch="", remote=remote, worktree_path=worktree_path
+        "commit_files", commit="", branch="", remote=remote
    )
    if blocked:
        return blocked
@@ -9406,7 +9358,7 @@ def gitea_commit_files(
        )
    # #735: forward explicit org/repo into shared anti-stomp preflight.
-    verify_preflight_purity(remote=remote, worktree_path=worktree_path, task="commit_files", org=org, repo=repo)
+    verify_preflight_purity(remote, task="commit_files", org=org, repo=repo)
    processed_files, source_proofs = _prepare_commit_payload_files(files)
    h, o, r = _resolve(remote, host, org, repo)
@@ -9683,96 +9635,6 @@ 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")
@@ -11269,6 +11131,9 @@ def _collect_branch_ownership_records(
    """
    records: list[dict] = []
    inventory_error = False
    # #855 AC4: expired/stale reviewer-lease records eligible for an explicit
    # reclaim decision, evaluated after the full ownership inventory is built.
    reviewer_reclaim_candidates: list[tuple[dict, bool | None]] = []
    target_branch = (branch or "").strip()
    if not target_branch:
        return {"records": records, "inventory_error": False}
@@ -11417,15 +11282,28 @@ def _collect_branch_ownership_records(
                else:
                    status = freshness_status
                    reclaim_allowed = False
-                records.append(
+                rec = _base_rec(
-                    _base_rec(
+                    category=category,
-                        category=category,
+                    status=status,
-                        status=status,
+                    reclaim_allowed=reclaim_allowed,
-                        reclaim_allowed=reclaim_allowed,
+                    role=role,
-                        role=role,
+                    host=lease_host or host_n or host,
                        host=lease_host or host_n or host,
                    )
                )
                records.append(rec)
                # #855 AC4: a reviewer lease that is expired/stale (its owner
                # gone) becomes a candidate for an explicit, fail-closed
                # reclaim decision made once the full inventory is known.
                if (
                    role == "reviewer"
                    and status
                    in branch_cleanup_guard._RECLAIMABLE_REVIEWER_STATUSES
                ):
                    owner_alive = (
                        fr.get("owner_pid_alive") if isinstance(fr, dict) else None
                    )
                    reviewer_reclaim_candidates.append(
                        (rec, owner_alive if isinstance(owner_alive, bool) else None)
                    )
    except Exception:
        # O1: fail closed on control-plane inventory errors.
        inventory_error = True
@@ -11496,6 +11374,44 @@ def _collect_branch_ownership_records(
            )
        )
    # #855 AC4: decide, explicitly and fail-closed, whether any expired/stale
    # reviewer lease may stop protecting an already-merged branch. This runs
    # only after the full ownership inventory is built, so a competing active
    # claimant (an active lease, author session, worktree binding, or active
    # reviewer comment lease) is visible. An inventory failure keeps every
    # reclaim candidate protective (reclaim_allowed stays False).
    if reviewer_reclaim_candidates and not inventory_error:
        pr_merged_state: bool | None = None
        if pr_number is not None and auth and base_api:
            try:
                pr_live = api_request(
                    "GET", f"{base_api}/pulls/{int(pr_number)}", auth
                )
                if isinstance(pr_live, dict) and pr_live:
                    pr_merged_state = bool(
                        pr_live.get("merged") or pr_live.get("merged_at")
                    )
            except Exception:
                # Unknown merged state fails closed (candidate stays protective).
                pr_merged_state = None
        for cand_rec, owner_alive in reviewer_reclaim_candidates:
            competing = any(
                other is not cand_rec
                and branch_cleanup_guard.is_active_ownership_status(
                    other.get("status")
                )
                for other in records
            )
            decision = branch_cleanup_guard.assess_expired_reviewer_lease_reclaim(
                role=str(cand_rec.get("role")),
                status=str(cand_rec.get("status")),
                pr_merged=pr_merged_state,
                owner_pid_alive=owner_alive,
                competing_active_claimant=competing,
            )
            cand_rec["reclaim_allowed"] = decision["reclaim_allowed"]
            cand_rec["reclaim_decision"] = decision["decision"]
    return {"records": records, "inventory_error": inventory_error}
@@ -11558,6 +11474,7 @@ def gitea_reconcile_merged_cleanups(
    dry_run: bool = True,
    execute_confirmed: bool = False,
    limit: int = 50,
    pr_number: int | None = None,
    remote: str = "dadeschools",
    host: str | None = None,
    org: str | None = None,
@@ -11568,7 +11485,11 @@ def gitea_reconcile_merged_cleanups(
    Args:
        dry_run: Defaults to True. When True, only builds the reconciliation report.
        execute_confirmed: Must be True when dry_run=False.
-        limit: Max number of closed PRs to inspect.
+        limit: Max number of closed PRs to inspect (batch mode only; ignored when
            ``pr_number`` is set).
        pr_number: Optional exact merged PR selector (#855). When set, only that
            PR is assessed/acted on (fail closed if missing, unmerged, or
            ambiguous). When omitted, existing batch behaviour is preserved.
        remote: Known Gitea instance ('dadeschools' or 'prgs').
        host: Override the Gitea host.
        org: Override the owner/organization.
@@ -11603,11 +11524,120 @@ def gitea_reconcile_merged_cleanups(
                "audit_phase": audit_reconciliation_mode.current_phase(),
            }
    # #855: optional exact PR pin. Fail closed before any inventory mutation.
    exact_pr: int | None = None
    if pr_number is not None:
        try:
            exact_pr = int(pr_number)
        except (TypeError, ValueError):
            return {
                "success": False,
                "performed": False,
                "executed": False,
                "dry_run": bool(dry_run),
                "selection_mode": "exact_pr",
                "selected_pr_number": pr_number,
                "reasons": [
                    f"pr_number={pr_number!r} is not a valid integer "
                    "(fail closed; no mutation)"
                ],
                "blocker_kind": "invalid_pr_number",
            }
        if exact_pr <= 0:
            return {
                "success": False,
                "performed": False,
                "executed": False,
                "dry_run": bool(dry_run),
                "selection_mode": "exact_pr",
                "selected_pr_number": exact_pr,
                "reasons": [
                    f"pr_number={exact_pr} must be a positive integer "
                    "(fail closed; no mutation)"
                ],
                "blocker_kind": "invalid_pr_number",
            }
    h, o, r = _resolve(remote, host, org, repo)
    auth = _auth(h)
    base = repo_api_url(h, o, r)
-    closed_prs = api_get_all(f"{base}/pulls?state=closed", auth, limit=limit)
+
-    open_prs = api_get_all(f"{base}/pulls?state=open", auth)
+    selection_mode = "batch"
    closed_prs: list[dict] = []
    open_prs: list[dict] = []
    if exact_pr is not None:
        selection_mode = "exact_pr"
        try:
            pr_live = api_request("GET", f"{base}/pulls/{exact_pr}", auth)
        except Exception as exc:
            return {
                "success": False,
                "performed": False,
                "executed": False,
                "dry_run": bool(dry_run),
                "selection_mode": selection_mode,
                "selected_pr_number": exact_pr,
                "reasons": [
                    f"PR #{exact_pr} could not be uniquely resolved "
                    f"(fail closed; no mutation): {_redact(str(exc))}"
                ],
                "blocker_kind": "pr_unresolvable",
            }
        if not isinstance(pr_live, dict) or not pr_live:
            return {
                "success": False,
                "performed": False,
                "executed": False,
                "dry_run": bool(dry_run),
                "selection_mode": selection_mode,
                "selected_pr_number": exact_pr,
                "reasons": [
                    f"PR #{exact_pr} could not be uniquely resolved "
                    "(empty response; fail closed; no mutation)"
                ],
                "blocker_kind": "pr_unresolvable",
            }
        live_number = pr_live.get("number")
        try:
            live_number_int = int(live_number) if live_number is not None else None
        except (TypeError, ValueError):
            live_number_int = None
        if live_number_int != exact_pr:
            return {
                "success": False,
                "performed": False,
                "executed": False,
                "dry_run": bool(dry_run),
                "selection_mode": selection_mode,
                "selected_pr_number": exact_pr,
                "reasons": [
                    f"PR #{exact_pr} resolution is ambiguous or mismatched "
                    f"(live number={live_number!r}; fail closed; no mutation)"
                ],
                "blocker_kind": "pr_ambiguous",
            }
        if not (pr_live.get("merged") or pr_live.get("merged_at")):
            return {
                "success": False,
                "performed": False,
                "executed": False,
                "dry_run": bool(dry_run),
                "selection_mode": selection_mode,
                "selected_pr_number": exact_pr,
                "reasons": [
                    f"PR #{exact_pr} is not merged "
                    "(exact-target cleanup requires a merged PR; "
                    "fail closed; no mutation)"
                ],
                "blocker_kind": "pr_not_merged",
            }
        closed_prs = [pr_live]
        # Exact mode still needs open heads for remote-delete safety gates.
        open_prs = api_get_all(f"{base}/pulls?state=open", auth)
    else:
        # Preserve historical call order (closed then open) for batch callers/tests.
        closed_prs = api_get_all(f"{base}/pulls?state=closed", auth, limit=limit)
        open_prs = api_get_all(f"{base}/pulls?state=open", auth)
    merged_closed: list[dict] = []
    remote_branch_exists: dict[str, bool] = {}
@@ -11634,6 +11664,13 @@ def gitea_reconcile_merged_cleanups(
    scratch_candidates = merged_cleanup_reconcile.discover_reviewer_scratch_worktrees(
        _canonical_local_git_root()
    )
    # #855: exact-target never inventories or mutates foreign PR scratch trees.
    if exact_pr is not None:
        scratch_candidates = [
            s
            for s in scratch_candidates
            if int(s.get("pr_number") or 0) == int(exact_pr)
        ]
    active_reviewer_leases: dict[int, bool] = {}
    pr_states: dict[int, dict] = {}
    for scratch in scratch_candidates:
@@ -11668,131 +11705,194 @@ def gitea_reconcile_merged_cleanups(
        active_reviewer_leases=active_reviewer_leases,
        pr_states=pr_states,
    )
    report["selection_mode"] = selection_mode
    if exact_pr is not None:
        report["selected_pr_number"] = exact_pr
        # Fail closed if exact pin somehow produced other or zero entries.
        entries = list(report.get("entries") or [])
        entry_numbers = []
        for entry in entries:
            try:
                entry_numbers.append(int(entry.get("pr_number")))
            except (TypeError, ValueError):
                entry_numbers.append(entry.get("pr_number"))
        if entry_numbers != [exact_pr]:
            return {
                "success": False,
                "performed": False,
                "executed": False,
                "dry_run": bool(dry_run),
                "selection_mode": selection_mode,
                "selected_pr_number": exact_pr,
                "reasons": [
                    f"exact PR #{exact_pr} selection produced unexpected "
                    f"candidate set {entry_numbers!r} "
                    "(fail closed; no mutation)"
                ],
                "blocker_kind": "exact_selection_mismatch",
                "entries": entries,
            }
    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] = []
    project_root = _canonical_local_git_root()
    def _ownership_records_for_branch(
        head_branch: str, pr_num_int: int | None
    ) -> list[dict]:
        ownership_bundle = _collect_branch_ownership_records(
            remote=remote,
            host=h,
            org=o,
            repo=r,
            branch=head_branch,
            pr_number=pr_num_int,
            project_root=project_root,
            auth=auth,
            base_api=base,
        )
        ownership_records = list(ownership_bundle.get("records") or [])
        if ownership_bundle.get("inventory_error"):
            ownership_records.append(
                {
                    "category": (
                        branch_cleanup_guard.OWNERSHIP_CATEGORY_INVENTORY_ERROR
                    ),
                    "status": "unknown",
                    "remote": remote,
                    "host": h,
                    "org": o,
                    "repo": r,
                    "branch": head_branch,
                    "reclaim_allowed": False,
                    "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,
            repo=r,
            branch=head_branch,
            host=h,
            records=ownership_records,
        )
        if ownership.get("block"):
            return {
                "action": "delete_remote_branch",
                "branch": head_branch,
                "success": False,
                "performed": False,
                "delete_acknowledged": False,
                "verified_absent": False,
                "blocker_kind": "active_branch_ownership",
                "reasons": ownership.get("reasons") or [],
                "blocking_categories": ownership.get("blocking_categories") or [],
                "after_worktree_removal": after_worktree_removal,
                "ownership_reassessed": after_worktree_removal,
            }
        encoded = urllib.parse.quote(head_branch, safe="")
        url = f"{base}/branches/{encoded}"
        with _audited(
            "delete_branch",
            host=h,
            remote=remote,
            org=o,
            repo=r,
            target_branch=head_branch,
            request_metadata={
                "branch": head_branch,
                "source": "reconcile_merged_cleanups",
                "ownership_checked": True,
                "after_worktree_removal": after_worktree_removal,
            },
        ):
            api_request("DELETE", url, auth)
        readback = _probe_remote_branch(h, o, r, auth, head_branch)
        readback_assessment = branch_cleanup_guard.assess_post_delete_readback(
            readback
        )
        verified = bool(readback_assessment.get("verified_absent"))
        return {
            "action": "delete_remote_branch",
            "branch": head_branch,
            "success": bool(readback_assessment.get("ok")),
            "performed": True,
            "delete_acknowledged": True,
            "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
-        if remote_assessment.get("safe_to_delete_remote"):
+        # #851 lifecycle: when the target worktree is independently safe, remove
-            import urllib.parse
+        # it first so worktree_binding ownership does not permanently strand
-
+        # both the worktree and the remote branch. Never skip worktree removal
-            pr_num = entry.get("pr_number")
+        # merely because remote delete would be blocked by that binding.
-            try:
+        # Ownership protection for remote delete remains fail-closed below.
-                pr_num_int = int(pr_num) if pr_num is not None else None
+        worktree_removed = False
            except (TypeError, ValueError):
                pr_num_int = None
            ownership_bundle = _collect_branch_ownership_records(
                remote=remote,
                host=h,
                org=o,
                repo=r,
                branch=head_branch,
                pr_number=pr_num_int,
                project_root=_canonical_local_git_root(),
                auth=auth,
                base_api=base,
            )
            ownership_records = list(ownership_bundle.get("records") or [])
            if ownership_bundle.get("inventory_error"):
                ownership_records.append(
                    {
                        "category": (
                            branch_cleanup_guard.OWNERSHIP_CATEGORY_INVENTORY_ERROR
                        ),
                        "status": "unknown",
                        "remote": remote,
                        "host": h,
                        "org": o,
                        "repo": r,
                        "branch": head_branch,
                        "reclaim_allowed": False,
                        "role": "inventory",
                    }
                )
            ownership = branch_cleanup_guard.assess_active_branch_ownership(
                remote=remote,
                org=o,
                repo=r,
                branch=head_branch,
                host=h,
                records=ownership_records,
            )
            if ownership.get("block"):
                actions.append(
                    {
                        "action": "delete_remote_branch",
                        "branch": head_branch,
                        "success": False,
                        "performed": False,
                        "delete_acknowledged": False,
                        "verified_absent": False,
                        "blocker_kind": "active_branch_ownership",
                        "reasons": ownership.get("reasons") or [],
                        "blocking_categories": ownership.get(
                            "blocking_categories"
                        )
                        or [],
                    }
                )
                continue
            encoded = urllib.parse.quote(head_branch, safe="")
            url = f"{base}/branches/{encoded}"
            with _audited(
                "delete_branch",
                host=h,
                remote=remote,
                org=o,
                repo=r,
                target_branch=head_branch,
                request_metadata={
                    "branch": head_branch,
                    "source": "reconcile_merged_cleanups",
                    "ownership_checked": True,
                },
            ):
                api_request("DELETE", url, auth)
            readback = _probe_remote_branch(h, o, r, auth, head_branch)
            readback_assessment = branch_cleanup_guard.assess_post_delete_readback(
                readback
            )
            verified = bool(readback_assessment.get("verified_absent"))
            actions.append(
                {
                    "action": "delete_remote_branch",
                    "branch": head_branch,
                    "success": bool(readback_assessment.get("ok")),
                    "performed": True,
                    "delete_acknowledged": True,
                    "verified_absent": verified,
                    "readback": readback_assessment.get("readback"),
                    "reasons": readback_assessment.get("reasons") or [],
                }
            )
        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})
@@ -19006,10 +19106,59 @@ def gitea_update_pr_branch_by_merge(
        prepared_verdict_head_sha=live_pr_head,
    )
-    return {
+    # #871: the remote head is now advanced; the durable linked-issue lock must
-        "success": True,
+    # be advanced with it, or a later dead-session recovery can never prove
    # ownership at the new head. This runs AFTER the successful remote update, so
    # a failure here is a *partial* lifecycle failure — the remote moved but the
    # durable state did not — and must never be reported as a full success.
    claimant = _work_lease_claimant(h)
    matched_issue = ownership.get("matched_issue")
    synced_at = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
    lock_refresh: dict = {
        "refreshed": False,
        "reasons": ["durable lock head refresh was not attempted"],
    }
    if new_head and matched_issue and source_branch and (wt or None):
        try:
            lock_refresh = issue_lock_store.apply_durable_lock_head_refresh(
                remote=remote,
                org=o,
                repo=r,
                issue_number=int(matched_issue),
                branch_name=source_branch,
                worktree_path=wt,
                pr_number=pr_number,
                identity=claimant.get("username"),
                profile=claimant.get("profile"),
                current_pid=os.getpid(),
                expected_old_head=live_pr_head,
                new_head=new_head,
                synced_at=synced_at,
                base_head=live_base_head,
            )
        except Exception as exc:
            lock_refresh = {
                "refreshed": False,
                "reasons": [
                    f"durable lock head refresh raised (fail closed): {_redact(str(exc))}"
                ],
            }
    else:
        lock_refresh = {
            "refreshed": False,
            "reasons": [
                "durable lock head refresh could not run: missing new head, "
                "linked issue, source branch, or worktree binding"
            ],
        }
    durable_refreshed = bool(lock_refresh.get("refreshed"))
    base_result = {
        "performed": True,
        "mutation_allowed": True,
        "durable_lock_refreshed": durable_refreshed,
        "durable_lock_refresh": lock_refresh,
        "fully_synchronized": durable_refreshed,
        "style": "merge",
        "force_push": False,
        "rebase": False,
@@ -19031,19 +19180,37 @@ def gitea_update_pr_branch_by_merge(
        "prepared_verdict_invalidated": transition.get(
            "prepared_verdict_invalidated"
        ),
        "recommended_next_action": transition.get(
            "recommended_next_action",
            pr_sync_status.ACTION_FRESH_REVIEW_REQUIRED,
        ),
        "transition": transition,
        "role_kind": role,
        "profile_name": profile.get("profile_name"),
        "worktree_path": wt or None,
        "reasons": list(transition.get("reasons") or []) + [
            "update-by-merge completed via native Gitea API (style=merge only)"
        ],
    }
    if durable_refreshed:
        base_result["success"] = True
        base_result["recommended_next_action"] = transition.get(
            "recommended_next_action",
            pr_sync_status.ACTION_FRESH_REVIEW_REQUIRED,
        )
        base_result["reasons"] = list(transition.get("reasons") or []) + [
            "update-by-merge completed via native Gitea API (style=merge only)",
            f"durable linked-issue lock #{matched_issue} head refreshed to "
            f"{new_head} (verified by read-after-write)",
        ]
        return base_result
    # Partial lifecycle failure: the remote advanced but the durable lock did
    # not. Do NOT report a fully successful synchronization (#871).
    base_result["success"] = False
    base_result["partial_lifecycle_failure"] = True
    base_result["recommended_next_action"] = pr_sync_status.ACTION_BLOCKED
    base_result["reasons"] = list(lock_refresh.get("reasons") or []) + [
        f"PARTIAL LIFECYCLE FAILURE: PR #{pr_number} remote head advanced to "
        f"{new_head} but the durable linked-issue lock head was not refreshed; "
        "the synchronization is NOT complete",
    ]
    return base_result
@mcp.tool()
 def gitea_assess_conflict_fix_push(
@@ -19705,12 +19872,6 @@ 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.
@@ -19727,14 +19888,13 @@ def gitea_resolve_task_capability(
        remote: Known remote instance name.
        host: Optional override for the Gitea host.
    """
    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_key not in TASK_MAP:
+    if task 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)
@@ -19780,8 +19940,8 @@ def gitea_resolve_task_capability(
            result["cleared_stale_denial"] = True
        return result
-    required_permission = task_capability_map.required_permission(task_key)
+    required_permission = task_capability_map.required_permission(task)
-    required_role = task_capability_map.required_role(task_key)
+    required_role = task_capability_map.required_role(task)
    role_exclusive_tasks = task_capability_map.ROLE_EXCLUSIVE_TASKS
    infra_assessment = role_session_router.assess_infra_stop(PROJECT_ROOT)
@@ -19967,10 +20127,7 @@ def gitea_resolve_task_capability(
    available_in_session = allowed_in_current_session
    runtime_stale_blocker = False
-    if (
+    if "PYTEST_CURRENT_TEST" not in os.environ or "GITEA_FORCE_MCP_RUNTIME_CHECK" in os.environ:
        "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
@@ -85,6 +85,12 @@ HEAD_RELATION_STRICT_DESCENDANT = "strict_descendant"
 # #772: an unpublished claim has no recorded head to compare against at all, so
 # its head is measured against the base the branch was cut from instead.
 HEAD_RELATION_DESCENDS_FROM_BASE = "descends_from_recorded_base"
 # #871: the remote/PR head advanced *past* the recorded head via a sanctioned
 # merge-based branch synchronization (``gitea_update_pr_branch_by_merge``) while
 # the local worktree stayed at the recorded head. This is the inverse of the
 # #768 descendant relation — here the *remote* strictly descends the local head,
 # and only because a base was merged into the branch, proven server-side.
 HEAD_RELATION_REMOTE_MERGE_SYNCED = "remote_merge_synced"
 # Which body of evidence a recovery was decided on (#772 AC10). These are not
 # interchangeable: a published claim proves ownership against a remote/PR head,
@@ -266,6 +272,70 @@ def _assess_base_descendancy(
    ]
 def _assess_remote_merge_synced(
    sync_provenance: Mapping[str, Any] | None,
    *,
    recorded_head: str,
    remote_head: str,
 ) -> tuple[bool, list[str]]:
    """Did ``remote_head`` advance past ``recorded_head`` via a sanctioned
    merge-based branch sync (#871)?
    ``sync_provenance`` is the server-side git observation from
    ``issue_lock_worktree.read_merge_sync_provenance``. Its own
    ``prior_head_sha`` / ``synced_head_sha`` are re-checked against the heads
    this assessment is actually reasoning about, so an observation taken for some
    other pair of commits — stale, mismatched, or hand-built — can never
    authorize recovery. This is the inverse of ``_assess_strict_descendant``: the
    recorded head is the ancestor and the *remote* head is the descendant, and it
    is accepted only because the remote head is a base-into-branch merge that
    preserved the branch mainline back to the recorded head.
    Returns ``(proven, notes)``. Notes name the exact missing element so a
    refused caller sees why, never a bare "unproven".
    """
    if not isinstance(sync_provenance, Mapping):
        return False, [
            "no server-derived merge-sync provenance observation was available; a "
            "remote head ahead of the recorded head cannot be accepted"
        ]
    probe_prior = _text(sync_provenance.get("prior_head_sha"))
    probe_synced = _text(sync_provenance.get("synced_head_sha"))
    if probe_prior != recorded_head or probe_synced != remote_head:
        return False, [
            f"merge-sync observation covers {probe_prior or 'unknown'} -> "
            f"{probe_synced or 'unknown'}, not the heads under assessment "
            f"({recorded_head} -> {remote_head})"
        ]
    if not sync_provenance.get("probe_ok"):
        return False, (
            list(sync_provenance.get("reasons") or [])
            or ["merge-sync provenance probe did not complete; provenance unproven"]
        )
    if not sync_provenance.get("prior_is_ancestor"):
        return False, [
            f"recorded head {recorded_head} is not an ancestor of remote head "
            f"{remote_head}; a rewritten or force-moved head cannot be recovered"
        ]
    if not sync_provenance.get("is_merge_sync"):
        return False, (
            list(sync_provenance.get("reasons") or [])
            or [
                f"remote head {remote_head} is not a sanctioned merge-based sync "
                f"of the base into the branch above {recorded_head}"
            ]
        )
    proof = _text(sync_provenance.get("proof")) or (
        f"{remote_head} merged the base into the branch above {recorded_head}"
    )
    return True, [
        f"remote head {remote_head} advanced past recorded head {recorded_head} "
        f"via a sanctioned merge-based branch sync ({proof})"
    ]
 def assess_dead_session_lock_recovery(
    existing_lock: Mapping[str, Any] | None,
    *,
@@ -290,6 +360,7 @@ def assess_dead_session_lock_recovery(
    remote_branch_exists: bool | None = None,
    recorded_base_sha: str | None = None,
    base_ancestry: Mapping[str, Any] | None = None,
    sync_provenance: Mapping[str, Any] | None = None,
 ) -> dict[str, Any]:
    """Decide whether a dead-session author lock may be natively recovered.
@@ -469,19 +540,39 @@ def assess_dead_session_lock_recovery(
                    head_relation = HEAD_RELATION_STRICT_DESCENDANT
                    ancestry_proof = notes[0] if notes else None
                else:
-                    reasons.append(
+                    # #871: the reverse relation — the remote head advanced past
-                        f"local head {local_head} does not match remote branch head "
+                    # the recorded/local head via a sanctioned merge-based branch
-                        f"{remote_head}"
+                    # sync while the local worktree stayed put. Accepted only on
                    # server-proven merge-sync provenance, never a caller claim.
                    synced, sync_notes = _assess_remote_merge_synced(
                        sync_provenance,
                        recorded_head=local_head,
                        remote_head=remote_head,
                    )
-                    reasons.extend(notes)
+                    if synced:
                        head_relation = HEAD_RELATION_REMOTE_MERGE_SYNCED
                        ancestry_proof = sync_notes[0] if sync_notes else None
                    else:
                        reasons.append(
                            f"local head {local_head} does not match remote branch "
                            f"head {remote_head}"
                        )
                        reasons.extend(notes)
                        reasons.extend(sync_notes)
    evidence["recorded_base"] = recorded_base or None
    evidence["local_head"] = local_head or None
    evidence["remote_head"] = remote_head or None
    # ``recorded_head`` is the head recovery is being measured against;
    # ``accepted_head`` is the head this recovery actually adopts. They differ
    # only in the descendant case, and downstream gates need both (#768 AC2/AC7).
    # #871: in the merge-sync case the branch/PR already carries the synced
    # remote head, so that is the head recovery adopts; the local worktree stays
    # at the ancestor recorded head.
    evidence["recorded_head"] = remote_head or None
-    evidence["accepted_head"] = local_head or None
+    if head_relation == HEAD_RELATION_REMOTE_MERGE_SYNCED:
        evidence["accepted_head"] = remote_head or None
    else:
        evidence["accepted_head"] = local_head or None
    evidence["head_relation"] = head_relation
    evidence["ancestry_proof"] = ancestry_proof
@@ -493,12 +584,17 @@ def assess_dead_session_lock_recovery(
        # contradictory; re-stating it as a head mismatch would only obscure why.
        if not unpublished and local_head and pr_head != local_head:
            # A descendant recovery has not been published yet, so the open PR
-            # legitimately still points at the recorded head. Any other
+            # legitimately still points at the recorded head. A merge-sync
-            # disagreement is a real mismatch.
+            # recovery's PR legitimately sits at the advanced remote head. Any
            # other disagreement is a real mismatch.
            if not (
                head_relation == HEAD_RELATION_STRICT_DESCENDANT
                and remote_head
                and pr_head == remote_head
            ) and not (
                head_relation == HEAD_RELATION_REMOTE_MERGE_SYNCED
                and remote_head
                and pr_head == remote_head
            ):
                reasons.append(
                    f"open PR #{pr_number} head {pr_head} does not match local head "
@@ -619,7 +715,11 @@ def assess_dead_session_lock_recovery(
        )
    if (
        head_relation
-        in (HEAD_RELATION_STRICT_DESCENDANT, HEAD_RELATION_DESCENDS_FROM_BASE)
+        in (
            HEAD_RELATION_STRICT_DESCENDANT,
            HEAD_RELATION_DESCENDS_FROM_BASE,
            HEAD_RELATION_REMOTE_MERGE_SYNCED,
        )
        and ancestry_proof
    ):
        proof.append(ancestry_proof)
@@ -697,6 +797,16 @@ def owning_pr_recovery_evidence(
            return None
        if accepted_head != local_head:
            return None
    elif relation == HEAD_RELATION_REMOTE_MERGE_SYNCED:
        # #871: the PR already sits at the advanced remote head; the local
        # worktree is the ancestor the merge preserved. The head the open PR
        # shows and the head recovery adopts are both the synced remote head.
        if not remote_head or pr_head != remote_head:
            return None
        if accepted_head != remote_head:
            return None
        if not local_head or local_head == remote_head:
            return None
    else:
        return None
    try:
@@ -765,6 +875,18 @@ def recovered_owning_pr_from_lock(
            return None
        if not accepted_head or accepted_head == recorded_head:
            return None
    elif relation == HEAD_RELATION_REMOTE_MERGE_SYNCED:
        # #871: PR sits at the advanced remote head, which is both the recorded
        # measured-against head and the adopted head; the local worktree is the
        # ancestor the merge preserved.
        remote_head = _text(record.get("remote_head"))
        local_head = _text(record.get("local_head"))
        if not remote_head or pr_head != remote_head:
            return None
        if accepted_head and accepted_head != remote_head:
            return None
        if not local_head or local_head == remote_head:
            return None
    else:
        return None
    try:
@@ -737,4 +737,308 @@ def format_lock_proof(
        parts.append("lock released")
    elif released is False:
        parts.append("lock retained")
-    return "; ".join(parts)
+    return "; ".join(parts)
 # ── #871: durable linked-issue lock head refresh after branch synchronization ──
 _FULL_SHA_RE = re.compile(r"^[0-9a-f]{40}$", re.IGNORECASE)
 # Provenance recorded on the lock when the head is refreshed by a sanctioned
 # merge-based branch synchronization (``gitea_update_pr_branch_by_merge``).
 LOCK_HEAD_REFRESH_PROVENANCE_MERGE_SYNC = "gitea_update_pr_branch_by_merge"
 def _norm_sha(value: Any) -> str | None:
    text = str(value or "").strip().lower()
    return text if _FULL_SHA_RE.match(text) else None
 def _lock_claimant_view(lock: dict[str, Any] | None) -> dict[str, Any]:
    if not isinstance(lock, dict):
        return {}
    claimant = lock.get("claimant")
    if not isinstance(claimant, dict):
        lease = lock.get("work_lease")
        claimant = lease.get("claimant") if isinstance(lease, dict) else None
    return dict(claimant) if isinstance(claimant, dict) else {}
 def assess_durable_lock_head_refresh(
    existing_lock: dict[str, Any] | None,
    *,
    remote: str,
    org: str,
    repo: str,
    issue_number: int,
    branch_name: str,
    worktree_path: str,
    pr_number: int | None,
    identity: str | None,
    profile: str | None,
    current_pid: int | None,
    expected_old_head: str | None,
    new_head: str | None,
    base_head: str | None = None,
 ) -> dict[str, Any]:
    """Fail-closed assessment for refreshing a durable lock's recorded head (#871).
    A successful ``gitea_update_pr_branch_by_merge`` advances the *remote* PR head
    but must also advance the durable linked-issue lock so a later dead-session
    recovery can prove ownership. This decides whether that refresh is permitted;
    it mutates nothing.
    Every element of durable ownership is re-verified against the persisted lock —
    repository, issue, branch, worktree, claimant identity/profile, and the live
    owning session — and the recorded head is compare-and-swapped: the lock's
    currently recorded synced head (if any) must equal ``expected_old_head``, so a
    lock whose head or provenance changed concurrently is never overwritten.
    """
    reasons: list[str] = []
    old = _norm_sha(expected_old_head)
    new = _norm_sha(new_head)
    evidence: dict[str, Any] = {
        "issue_number": issue_number,
        "branch_name": branch_name,
        "worktree_path": worktree_path,
        "pr_number": pr_number,
        "expected_old_head": old,
        "new_head": new,
        "base_head": _norm_sha(base_head),
    }
    if not isinstance(existing_lock, dict) or not existing_lock:
        reasons.append("no durable lock exists for this issue; nothing to refresh")
        return {"allowed": False, "reasons": reasons, "evidence": evidence,
                "expected_generation": 0}
    lock = dict(existing_lock)
    evidence["current_generation"] = lock_generation(lock)
    if lock.get("issue_number") != issue_number:
        reasons.append(
            f"durable lock targets issue #{lock.get('issue_number')}, not "
            f"#{issue_number}; refusing head refresh"
        )
    for field, expected in (("remote", remote), ("org", org), ("repo", repo)):
        actual = str(lock.get(field) or "").strip()
        if actual != str(expected or "").strip():
            reasons.append(
                f"lock {field} '{actual}' does not match requested "
                f"'{str(expected or '').strip()}'"
            )
    locked_branch = str(lock.get("branch_name") or "").strip()
    if locked_branch != str(branch_name or "").strip():
        reasons.append(
            f"lock branch '{locked_branch}' does not match requested "
            f"'{str(branch_name or '').strip()}'"
        )
    locked_worktree = str(lock.get("worktree_path") or "").strip()
    try:
        same_wt = bool(locked_worktree) and bool(worktree_path) and (
            os.path.realpath(locked_worktree) == os.path.realpath(worktree_path)
        )
    except OSError:
        same_wt = locked_worktree == (worktree_path or "")
    if not same_wt:
        reasons.append(
            f"lock worktree '{locked_worktree}' does not match declared "
            f"'{str(worktree_path or '').strip()}'"
        )
    claimant = _lock_claimant_view(lock)
    locked_identity = str(claimant.get("username") or "").strip()
    locked_profile = str(claimant.get("profile") or "").strip()
    if not locked_identity or not locked_profile:
        reasons.append(
            "durable lock does not record a claimant identity/profile; "
            "ownership could not be proven for head refresh"
        )
    if not str(identity or "").strip() or not str(profile or "").strip():
        reasons.append(
            "active session identity/profile is unknown; ownership could not be "
            "proven for head refresh"
        )
    if locked_identity and str(identity or "").strip() and locked_identity != str(identity).strip():
        reasons.append(
            f"lock claimant '{locked_identity}' does not match active identity "
            f"'{str(identity).strip()}'"
        )
    if locked_profile and str(profile or "").strip() and locked_profile != str(profile).strip():
        reasons.append(
            f"lock profile '{locked_profile}' does not match active profile "
            f"'{str(profile).strip()}'"
        )
    # The refresh is written by the LIVE owning author session. A refresh is not
    # a recovery: the current process must be the recorded owner.
    recorded_pid = lock.get("session_pid")
    if recorded_pid is None:
        recorded_pid = lock.get("pid")
    evidence["recorded_pid"] = recorded_pid
    evidence["current_pid"] = current_pid
    if current_pid is None:
        reasons.append("current session pid is unknown; cannot prove live ownership")
    else:
        try:
            if recorded_pid is None or int(recorded_pid) != int(current_pid):
                reasons.append(
                    f"durable lock is owned by pid {recorded_pid}, not the current "
                    f"session pid {current_pid}; head refresh requires the live owner"
                )
        except (TypeError, ValueError):
            reasons.append(
                "durable lock owner pid is malformed; cannot prove live ownership"
            )
    if not old:
        reasons.append("expected_old_head is not a full 40-char hex SHA (fail closed)")
    if not new:
        reasons.append("new_head is not a full 40-char hex SHA (fail closed)")
    if old and new and old == new:
        reasons.append(
            "new head equals the expected old head; a sync must advance the head"
        )
    # Compare-and-swap on the recorded head: if the lock already records a synced
    # head it must be exactly the expected old head, else another sync moved it.
    recorded_synced = _norm_sha(lock.get("synced_pr_head"))
    evidence["recorded_synced_pr_head"] = recorded_synced
    if recorded_synced is not None and old is not None and recorded_synced != old:
        reasons.append(
            f"durable lock already records synced head {recorded_synced}, not the "
            f"expected old head {old}; a concurrent sync changed it (CAS fail closed)"
        )
    if reasons:
        return {"allowed": False, "reasons": reasons, "evidence": evidence,
                "expected_generation": lock_generation(lock)}
    return {
        "allowed": True,
        "reasons": [
            f"durable lock for issue #{issue_number} branch '{locked_branch}' is "
            f"owned by the live session; refresh recorded head {old} -> {new}"
        ],
        "evidence": evidence,
        "expected_generation": lock_generation(lock),
    }
 def apply_durable_lock_head_refresh(
    *,
    remote: str,
    org: str,
    repo: str,
    issue_number: int,
    branch_name: str,
    worktree_path: str,
    pr_number: int | None,
    identity: str | None,
    profile: str | None,
    current_pid: int | None,
    expected_old_head: str | None,
    new_head: str | None,
    synced_at: str,
    base_head: str | None = None,
    provenance: str = LOCK_HEAD_REFRESH_PROVENANCE_MERGE_SYNC,
    lock_dir: str | None = None,
 ) -> dict[str, Any]:
    """CAS-refresh the durable lock's recorded head after a branch sync (#871).
    Reads the durable lock from disk, re-asserts ownership via
    ``assess_durable_lock_head_refresh``, and — only when permitted — writes the
    new synced head through ``bind_session_lock`` with a generation compare-and-
    swap. Then re-reads the lock and proves it records the complete new head
    (read-after-write). Any failure at any step returns ``refreshed=False`` with
    reasons; the caller must treat that as a partial lifecycle failure and never
    report a fully successful synchronization.
    """
    existing = load_issue_lock(
        remote=remote, org=org, repo=repo, issue_number=issue_number, lock_dir=lock_dir
    )
    assessment = assess_durable_lock_head_refresh(
        existing,
        remote=remote,
        org=org,
        repo=repo,
        issue_number=issue_number,
        branch_name=branch_name,
        worktree_path=worktree_path,
        pr_number=pr_number,
        identity=identity,
        profile=profile,
        current_pid=current_pid,
        expected_old_head=expected_old_head,
        new_head=new_head,
        base_head=base_head,
    )
    result: dict[str, Any] = {
        "refreshed": False,
        "read_after_write_ok": False,
        "prior_head": _norm_sha(expected_old_head),
        "new_head": _norm_sha(new_head),
        "reasons": list(assessment.get("reasons") or []),
        "evidence": assessment.get("evidence"),
    }
    if not assessment.get("allowed"):
        return result
    new = _norm_sha(new_head)
    old = _norm_sha(expected_old_head)
    record = dict(existing or {})
    sync_block = {
        "last_synced_pr_head": new,
        "prior_pr_head": old,
        "base_head": _norm_sha(base_head),
        "pr_number": pr_number,
        "provenance": provenance,
        "synced_at": synced_at,
        "synced_by_pid": current_pid,
        "synced_by": {
            "username": str(identity or "").strip() or None,
            "profile": str(profile or "").strip() or None,
        },
    }
    record["synced_pr_head"] = new
    record["branch_sync"] = sync_block
    history = record.get("branch_sync_history")
    if not isinstance(history, list):
        history = []
    history = list(history)
    history.append(sync_block)
    record["branch_sync_history"] = history
    try:
        bind_session_lock(
            record,
            lock_dir=lock_dir,
            expected_generation=assessment.get("expected_generation"),
            renewal_sanctioned=True,
        )
    except Exception as exc:  # CAS miss or write failure — partial lifecycle failure
        result["reasons"].append(
            f"durable lock head refresh write failed (fail closed): {exc}"
        )
        return result
    after = load_issue_lock(
        remote=remote, org=org, repo=repo, issue_number=issue_number, lock_dir=lock_dir
    )
    after_head = _norm_sha((after or {}).get("synced_pr_head"))
    result["lock_generation_after"] = lock_generation(after)
    if after_head == new and new is not None:
        result["refreshed"] = True
        result["read_after_write_ok"] = True
        result["reasons"].append(
            f"durable lock recorded head refreshed to {new} and verified by "
            "read-after-write"
        )
    else:
        result["reasons"].append(
            "read-after-write verification failed: durable lock does not record "
            f"the new head {new} (found {after_head}); partial lifecycle failure"
        )
    return result
@@ -145,6 +145,175 @@ def read_head_ancestry(
    return result
 def read_merge_sync_provenance(
    worktree_path: str,
    *,
    prior_head_sha: str | None,
    synced_head_sha: str | None,
 ) -> dict:
    """Observe whether ``synced_head_sha`` is a sanctioned merge-based branch sync
    that advanced the PR branch past ``prior_head_sha`` (#871).
    ``gitea_update_pr_branch_by_merge`` advances a PR branch by merging the base
    branch *into* the branch (``POST /pulls/{n}/update?style=merge``). The result
    is a merge commit ``M`` on the branch whose **first** parent is the prior
    branch head and whose second parent is the base tip. When the owning session
    then dies without the durable lock's recorded head being refreshed, the local
    worktree still sits at ``prior_head_sha`` while the live PR head is ``M``.
    Recovering that drift safely requires proving the remote head is *exactly*
    such a merge-sync — not a rewrite, rebase, force-push, or an unrelated
    commit. This is that server-side observation. It reports facts only; the
    disposition lives in ``issue_lock_recovery``. Every field is read from git in
    the declared worktree — nothing is supplied by, or reachable from, an MCP
    caller (#871).
    Provenance is proven only when ALL hold:
    * both commits are present (a rewritten/force-moved prior head leaves the
      object graph and fails closed);
    * ``prior_head_sha`` is a strict ancestor of ``synced_head_sha`` (the branch
      history is preserved, never replaced);
    * ``synced_head_sha`` is a merge commit (two or more parents), i.e. a base
      merged in — a plain fast-forward of new direct commits is not a sync;
    * ``prior_head_sha`` is an ancestor of the merge's **first** parent, so the
      branch mainline (first-parent lineage) still reaches the prior head — a
      rebase/force-push that re-authored the branch side fails this.
    """
    path = (worktree_path or "").strip()
    prior = (prior_head_sha or "").strip()
    synced = (synced_head_sha or "").strip()
    result: dict = {
        "prior_head_sha": prior or None,
        "synced_head_sha": synced or None,
        "probe_ok": False,
        "prior_present": False,
        "synced_present": False,
        "prior_is_ancestor": False,
        "synced_is_merge": False,
        "first_parent_reaches_prior": False,
        "is_merge_sync": False,
        "first_parent_sha": None,
        "parent_count": None,
        "proof": None,
        "reasons": [],
    }
    if not path or not prior or not synced:
        result["reasons"].append(
            "merge-sync provenance probe requires a worktree path and both "
            "commit SHAs"
        )
        return result
    if prior == synced:
        result["reasons"].append(
            "prior and synced heads are identical; no branch sync occurred"
        )
        return result
    def _present(sha: str) -> bool:
        res = subprocess.run(
            ["git", "-C", path, "rev-parse", "--verify", "--quiet", f"{sha}^{{commit}}"],
            capture_output=True,
            text=True,
            check=False,
        )
        return res.returncode == 0
    def _is_ancestor(ancestor: str, descendant: str) -> bool | None:
        res = subprocess.run(
            ["git", "-C", path, "merge-base", "--is-ancestor", ancestor, descendant],
            capture_output=True,
            text=True,
            check=False,
        )
        if res.returncode == 0:
            return True
        if res.returncode == 1:
            return False
        return None  # failed probe — never a silent "no"
    try:
        result["prior_present"] = _present(prior)
        result["synced_present"] = _present(synced)
    except OSError as exc:  # git unavailable — fail closed, never assume
        result["reasons"].append(f"merge-sync provenance probe could not run: {exc}")
        return result
    if not result["prior_present"]:
        result["reasons"].append(
            f"prior head {prior} is not reachable in '{path}'; history may have "
            "been rewritten or force-moved"
        )
    if not result["synced_present"]:
        result["reasons"].append(
            f"synced head {synced} is not reachable in '{path}'"
        )
    if not (result["prior_present"] and result["synced_present"]):
        return result
    ancestor = _is_ancestor(prior, synced)
    if ancestor is None:
        result["reasons"].append(
            "ancestry probe failed; merge-sync provenance unproven"
        )
        return result
    result["prior_is_ancestor"] = bool(ancestor)
    if not ancestor:
        result["reasons"].append(
            f"prior head {prior} is not an ancestor of synced head {synced}; "
            "the branch history was not preserved (not a merge-based sync)"
        )
        return result
    parents_res = subprocess.run(
        ["git", "-C", path, "rev-list", "--parents", "-n", "1", synced],
        capture_output=True,
        text=True,
        check=False,
    )
    if parents_res.returncode != 0:
        result["reasons"].append(
            f"could not read parents of {synced}; merge-sync provenance unproven"
        )
        return result
    tokens = (parents_res.stdout or "").split()
    # tokens[0] is the commit itself; the rest are its parents.
    parents = tokens[1:]
    result["parent_count"] = len(parents)
    result["synced_is_merge"] = len(parents) >= 2
    if not result["synced_is_merge"]:
        result["probe_ok"] = True
        result["reasons"].append(
            f"synced head {synced} has {len(parents)} parent(s); a merge-based "
            "branch sync produces a merge commit (two or more parents)"
        )
        return result
    first_parent = parents[0]
    result["first_parent_sha"] = first_parent
    fp_reaches = _is_ancestor(prior, first_parent) if prior != first_parent else True
    if fp_reaches is None:
        result["reasons"].append(
            "first-parent ancestry probe failed; merge-sync provenance unproven"
        )
        return result
    result["first_parent_reaches_prior"] = bool(fp_reaches)
    result["probe_ok"] = True
    if not fp_reaches:
        result["reasons"].append(
            f"merge first parent {first_parent} does not reach prior head "
            f"{prior}; the branch mainline was re-authored (not a sanctioned sync)"
        )
        return result
    result["is_merge_sync"] = True
    result["proof"] = (
        f"synced head {synced} is a merge commit (parents={len(parents)}) whose "
        f"first-parent lineage reaches prior head {prior}; base merged into branch"
    )
    return result
 def read_recorded_base(
    worktree_path: str,
    *,
@@ -73,8 +73,6 @@ 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",
@@ -69,14 +69,6 @@ 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",
@@ -496,8 +488,6 @@ 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"),
 })
@@ -544,8 +534,6 @@ 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",
@@ -571,7 +559,6 @@ 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",
 }
@@ -1,483 +0,0 @@
 """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
 from unittest import mock
 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,
        owner_session="session-concurrent-test",
        active_identity="jcwalker3",
        active_profile="prgs-author",
    )
 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,
            owner_session="session-test-1234",
        )
        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, journal_dir=self.lock_dir)
        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,
            owner_session="session-test-1234",
        )
        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,
            owner_session="session-test-1234",
        )
        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,
            owner_session="session-test-1234",
        )
        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,
            owner_session="session-test-1234",
        )
        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,
            owner_session="session-test-1234",
        )
        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, journal_dir=self.lock_dir)
        # Now resume/replay the transition but simulate lock binding failure during Phase 6
        with 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,
                owner_session="session-test-1234",
            )
        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 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,
                owner_session="session-test-1234",
            )
        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,
            owner_session="session-test-1234",
        )
        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,
            owner_session="session-test-1234",
        )
        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,
            owner_session="session-test-1234",
        )
        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_missing_owner_session_refusal(self):
        """Finding D: Missing owner_session context fails closed with typed refusal and zero mutation."""
        res = author_issue_bootstrap.bootstrap_author_issue_worktree(
            issue_number=850,
            canonical_repo_root=self.repo_dir,
            owner_session=None,
            lock_dir=self.lock_dir,
        )
        self.assertFalse(res["success"])
        self.assertEqual(res.get("reason_code"), "missing_owner_session")
        self.assertIn("exact_next_action", res)
    def test_symlink_lock_file_refusal(self):
        """Finding C: BootstrapTransitionLock refuses to follow symlinks."""
        key = "test_symlink_lock_key"
        safe_key = "".join(c if c.isalnum() or c in ("-", "_", ".") else "_" for c in key)
        lock_path = os.path.join(self.lock_dir, f"{safe_key}.lock")
        target_file = os.path.join(self.tmp_dir, "fake_target")
        with open(target_file, "w") as f:
            f.write("target")
        os.symlink(target_file, lock_path)
        with self.assertRaises(RuntimeError) as ctx:
            with author_issue_bootstrap.BootstrapTransitionLock(key, journal_dir=self.lock_dir):
                pass
        self.assertIn("symlink", str(ctx.exception).lower())
    def test_lock_directory_escape_refusal(self):
        """Finding C: BootstrapTransitionLock refuses keys that escape lock directory."""
        with mock.patch("os.path.abspath", return_value="/tmp/outside/evil_key.lock"):
            with self.assertRaises(RuntimeError) as ctx:
                author_issue_bootstrap.BootstrapTransitionLock("key", journal_dir=self.lock_dir)
        self.assertIn("escapes", str(ctx.exception).lower())
    def test_missing_active_identity_refusal(self):
        """F-5: Missing active_identity parameter fails closed."""
        res = author_issue_bootstrap.bootstrap_author_issue_worktree(
            issue_number=850,
            canonical_repo_root=self.repo_dir,
            owner_session="session-test-1234",
            active_identity=None,
            active_profile="prgs-author",
            lock_dir=self.lock_dir,
        )
        self.assertFalse(res["success"])
        self.assertEqual(res.get("reason_code"), "missing_active_identity")
    def test_missing_active_profile_refusal(self):
        """F-5: Missing active_profile parameter fails closed."""
        res = author_issue_bootstrap.bootstrap_author_issue_worktree(
            issue_number=850,
            canonical_repo_root=self.repo_dir,
            owner_session="session-test-1234",
            active_identity="jcwalker3",
            active_profile=None,
            lock_dir=self.lock_dir,
        )
        self.assertFalse(res["success"])
        self.assertEqual(res.get("reason_code"), "missing_active_profile")
    def test_dirty_worktree_preserved_during_recovery(self):
        """F-4: Compensating recovery does not delete dirty worktree."""
        branch = "fix/issue-850-rec-dirty"
        wt_path = os.path.join(self.branches_dir, "fix-issue-850-rec-dirty")
        subprocess.run(["git", "-C", self.repo_dir, "worktree", "add", "-b", branch, wt_path], check=True, capture_output=True)
        dirty_file = os.path.join(wt_path, "dirty.txt")
        with open(dirty_file, "w") as f:
            f.write("uncommitted work")
        journal = {
            "idempotency_key": "test_dirty_rec",
            "issue_number": 850,
            "branch_name": branch,
            "worktree_path": wt_path,
            "artifacts_created": {
                "worktree_dir_created": True,
                "worktree_registered": True,
            },
            "failure_reason": "test dirty recovery",
        }
        rec = author_issue_bootstrap.run_compensating_recovery(journal, self.repo_dir, journal_dir=self.lock_dir)
        self.assertTrue(os.path.exists(wt_path))
        self.assertIn(f"worktree_path_preserved_dirty:{wt_path}", rec["rolled_back"])
    def test_branch_with_commits_preserved_during_recovery(self):
        """F-4: Compensating recovery does not delete branch with author commits."""
        branch = "fix/issue-850-rec-commits"
        subprocess.run(["git", "-C", self.repo_dir, "branch", branch, self.master_sha], check=True, capture_output=True)
        # Add a commit on the branch
        wt_path = os.path.join(self.branches_dir, "fix-issue-850-rec-commits")
        subprocess.run(["git", "-C", self.repo_dir, "worktree", "add", wt_path, branch], check=True, capture_output=True)
        cfile = os.path.join(wt_path, "commit.txt")
        with open(cfile, "w") as f:
            f.write("author commit")
        subprocess.run(["git", "-C", wt_path, "add", "commit.txt"], check=True, capture_output=True)
        subprocess.run(["git", "-C", wt_path, "commit", "-m", "author commit"], check=True, capture_output=True)
        subprocess.run(["git", "-C", self.repo_dir, "worktree", "remove", "--force", wt_path], check=True, capture_output=True)
        journal = {
            "idempotency_key": "test_commits_rec",
            "issue_number": 850,
            "branch_name": branch,
            "resolved_base_sha": self.master_sha,
            "artifacts_created": {
                "branch_created": True,
            },
            "failure_reason": "test commit branch recovery",
        }
        rec = author_issue_bootstrap.run_compensating_recovery(journal, self.repo_dir, journal_dir=self.lock_dir)
        branch_check = subprocess.run(["git", "-C", self.repo_dir, "rev-parse", "--verify", branch], capture_output=True, text=True, check=False)
        self.assertEqual(branch_check.returncode, 0, "Branch with commits was deleted!")
        self.assertIn(f"branch_preserved_commits:{branch}", rec["rolled_back"])
    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()
@@ -28,21 +28,6 @@ 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"
@@ -1639,6 +1639,358 @@ class TestSecondRemediationIntegration(unittest.TestCase):
        self.assertTrue(ownership_calls)
 class TestIssue855ExactPrSelector(unittest.TestCase):
    """#855: exact pr_number pin for reconcile_merged_cleanups (#851 lifecycle)."""
    def setUp(self):
        self._remotes = patch.dict(
            mcp_server.REMOTES,
            {
                "prgs": {
                    "host": "gitea.example.com",
                    "org": "Scaled-Tech-Consulting",
                    "repo": "Gitea-Tools",
                }
            },
        )
        self._remotes.start()
        patch("gitea_audit.audit_enabled", return_value=False).start()
        self.mock_api = patch("mcp_server.api_request").start()
        self.mock_all = patch("mcp_server.api_get_all", return_value=[]).start()
        patch("mcp_server.get_auth_header", return_value=FAKE_AUTH).start()
        patch(
            "mcp_server.merged_cleanup_reconcile.is_head_ancestor_of_ref",
            return_value=True,
        ).start()
        patch(
            "mcp_server.get_profile",
            return_value=dict(RECONCILER_WITH_DELETE),
        ).start()
        patch(
            "mcp_server._profile_operation_gate",
            return_value=[],
        ).start()
        patch(
            "mcp_server._collect_branch_ownership_records",
            return_value={"records": [], "inventory_error": False},
        ).start()
        patch(
            "mcp_server.merged_cleanup_reconcile.discover_reviewer_scratch_worktrees",
            return_value=[],
        ).start()
        patch("mcp_server.verify_preflight_purity", return_value=None).start()
        patch(
            "mcp_server.audit_reconciliation_mode.check_cleanup_execution_allowed",
            return_value=(True, []),
        ).start()
    def tearDown(self):
        patch.stopall()
    def _merged_pr(self, number, branch, sha="c" * 40):
        return {
            "number": number,
            "title": f"PR {number}",
            "body": f"Closes #{number - 4}",
            "merged": True,
            "merged_at": "2026-07-23T12:00:00Z",
            "merge_commit_sha": "f" * 40,
            "state": "closed",
            "head": {"ref": branch, "sha": sha},
            "base": {"ref": "master"},
        }
    def test_exact_pr_848_ignores_newer_852_in_batch_queue(self):
        """pr_number=848 selects only #848 even when #852 is newer/first."""
        from mcp_server import gitea_reconcile_merged_cleanups
        pr_848 = self._merged_pr(
            848, "fix/issue-844-exclude-epic-containers", sha="c3f282ba" + "0" * 32
        )
        # Closed list would rank #852 first in batch mode; exact pin must ignore it.
        closed_batch = [
            self._merged_pr(852, "fix/issue-851-cleanup-worktree-before-remote-delete"),
            pr_848,
            self._merged_pr(849, "fix/issue-849-other"),
            self._merged_pr(846, "fix/issue-846-other"),
            self._merged_pr(845, "fix/issue-845-other"),
        ]
        batch_fetch_calls = []
        def fake_api(method, url, *args, **kwargs):
            if method == "GET" and url.rstrip("/").endswith("/pulls/848"):
                return dict(pr_848)
            if method == "GET" and "/pulls/" in url:
                raise AssertionError(f"unexpected PR fetch: {url}")
            if method == "GET" and "/branches/" in url:
                return {"name": "present"}
            return {}
        def fake_all(url, auth, limit=None):
            batch_fetch_calls.append((url, limit))
            if "state=open" in url:
                return []
            if "state=closed" in url:
                # Exact mode must not use the closed batch list.
                raise AssertionError(
                    "exact pr_number mode must not page closed PRs: " + url
                )
            return []
        self.mock_api.side_effect = fake_api
        self.mock_all.side_effect = fake_all
        patch(
            "mcp_server._remote_branch_exists",
            return_value=True,
        ).start()
        patch(
            "mcp_server.merged_cleanup_reconcile.build_reconciliation_report",
            side_effect=lambda **kwargs: {
                "entries": [
                    {
                        "pr_number": int(pr["number"]),
                        "head_branch": (pr.get("head") or {}).get("ref"),
                        "issue_number": 844,
                        "remote_branch": {
                            "safe_to_delete_remote": True,
                            "head_branch": (pr.get("head") or {}).get("ref"),
                        },
                        "local_worktree": {
                            "safe_to_remove_worktree": True,
                            "worktree_path": (
                                "/tmp/branches/fix-issue-844-exclude-epic-containers"
                            ),
                        },
                        "planned_execution_order": (
                            mcp_server.merged_cleanup_reconcile.plan_cleanup_execution_order(
                                remote_assessment={"safe_to_delete_remote": True},
                                local_assessment={"safe_to_remove_worktree": True},
                            )
                        ),
                    }
                    for pr in kwargs.get("closed_prs") or []
                    if pr.get("merged_at") or pr.get("merged")
                ],
                "reviewer_scratch_entries": [],
                "merged_pr_count": len(kwargs.get("closed_prs") or []),
            },
        ).start()
        res = gitea_reconcile_merged_cleanups(
            dry_run=True,
            pr_number=848,
            remote="prgs",
            org="Scaled-Tech-Consulting",
            repo="Gitea-Tools",
        )
        self.assertTrue(res.get("success"))
        self.assertFalse(res.get("performed"))
        self.assertEqual(res.get("selection_mode"), "exact_pr")
        self.assertEqual(res.get("selected_pr_number"), 848)
        entries = res.get("entries") or []
        self.assertEqual(len(entries), 1, entries)
        self.assertEqual(entries[0].get("pr_number"), 848)
        self.assertEqual(
            entries[0].get("head_branch"),
            "fix/issue-844-exclude-epic-containers",
        )
        # No other PR appears in plan.
        self.assertEqual(list((res.get("planned_execution_orders") or {}).keys()), ["848"])
        plan = (res.get("planned_execution_orders") or {}).get("848") or []
        actions = [s.get("action") for s in plan]
        self.assertEqual(
            actions,
            [
                "remove_local_worktree",
                "reassess_branch_ownership",
                "delete_remote_branch",
            ],
        )
        # Prove we never scanned the multi-PR closed batch.
        self.assertFalse(any("state=closed" in (u or "") for u, _ in batch_fetch_calls))
        # closed_batch fixture must remain unused (sanity).
        self.assertEqual(closed_batch[0]["number"], 852)
    def test_exact_pr_execute_only_mutates_selected_pr(self):
        """Execute with pr_number must never touch #845/#846/#849/#852."""
        from mcp_server import gitea_reconcile_merged_cleanups
        pr_848 = self._merged_pr(848, "fix/issue-844-exclude-epic-containers")
        worktree_path = "/tmp/branches/fix-issue-844-exclude-epic-containers"
        remove_calls = []
        delete_api_calls = []
        ownership_branches = []
        def fake_api(method, url, *args, **kwargs):
            if method == "GET" and url.rstrip("/").endswith("/pulls/848"):
                return dict(pr_848)
            if method == "DELETE":
                delete_api_calls.append(url)
                # Forbid foreign PR branch deletion by URL content.
                for forbidden in ("845", "846", "849", "852"):
                    self.assertNotIn(forbidden, url)
            return {}
        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_path}",
                "worktree_path": worktree_path,
            }
        def fake_collect(**kwargs):
            ownership_branches.append(kwargs.get("branch"))
            return {"records": [], "inventory_error": False}
        def fake_probe(h, o, r, auth, br):
            return guard.classify_branch_readback_http_status(
                404, not_found_scope=guard.NOT_FOUND_SCOPE_BRANCH
            )
        self.mock_api.side_effect = fake_api
        self.mock_all.side_effect = lambda url, auth, limit=None: []
        patch("mcp_server._remote_branch_exists", return_value=True).start()
        patch(
            "mcp_server.merged_cleanup_reconcile.build_reconciliation_report",
            return_value={
                "entries": [
                    {
                        "pr_number": 848,
                        "head_branch": "fix/issue-844-exclude-epic-containers",
                        "remote_branch": {"safe_to_delete_remote": True},
                        "local_worktree": {
                            "safe_to_remove_worktree": True,
                            "worktree_path": worktree_path,
                        },
                        "planned_execution_order": [
                            {"action": "remove_local_worktree", "phase": 1},
                            {"action": "reassess_branch_ownership", "phase": 2},
                            {"action": "delete_remote_branch", "phase": 3},
                        ],
                    }
                ],
                "reviewer_scratch_entries": [
                    # Foreign scratch must be filtered before report execute loop;
                    # if present here it would still be a test failure if acted on.
                ],
                "merged_pr_count": 1,
            },
        ).start()
        patch(
            "mcp_server.merged_cleanup_reconcile.remove_local_worktree",
            side_effect=fake_remove,
        ).start()
        patch(
            "mcp_server._collect_branch_ownership_records",
            side_effect=fake_collect,
        ).start()
        patch("mcp_server._probe_remote_branch", side_effect=fake_probe).start()
        res = gitea_reconcile_merged_cleanups(
            dry_run=False,
            execute_confirmed=True,
            pr_number=848,
            remote="prgs",
            org="Scaled-Tech-Consulting",
            repo="Gitea-Tools",
        )
        self.assertTrue(res.get("performed") or res.get("executed"))
        self.assertEqual(res.get("selection_mode"), "exact_pr")
        self.assertEqual(res.get("selected_pr_number"), 848)
        actions = res.get("actions") or []
        pr_numbers_touched = {
            a.get("pr_number") for a in actions if a.get("pr_number") is not None
        }
        self.assertTrue(pr_numbers_touched.issubset({None, 848}) or not pr_numbers_touched)
        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.assertEqual(remove_calls[0]["branch"], "fix/issue-844-exclude-epic-containers")
        self.assertEqual(len(deletes), 1)
        self.assertTrue(deletes[0].get("success"))
        self.assertTrue(deletes[0].get("after_worktree_removal"))
        self.assertEqual(len(delete_api_calls), 1)
        self.assertEqual(
            ownership_branches, ["fix/issue-844-exclude-epic-containers"]
        )
    def test_exact_pr_unknown_fails_closed_without_mutation(self):
        from mcp_server import gitea_reconcile_merged_cleanups
        def fake_api(method, url, *args, **kwargs):
            if method == "GET" and "/pulls/99999" in url:
                raise RuntimeError("HTTP 404 Not Found")
            raise AssertionError(f"unexpected API call {method} {url}")
        self.mock_api.side_effect = fake_api
        res = gitea_reconcile_merged_cleanups(
            dry_run=True,
            pr_number=99999,
            remote="prgs",
        )
        self.assertFalse(res.get("success"))
        self.assertFalse(res.get("performed"))
        self.assertEqual(res.get("blocker_kind"), "pr_unresolvable")
        self.assertIn("99999", " ".join(res.get("reasons") or []))
    def test_exact_pr_not_merged_fails_closed(self):
        from mcp_server import gitea_reconcile_merged_cleanups
        def fake_api(method, url, *args, **kwargs):
            if method == "GET" and url.rstrip("/").endswith("/pulls/900"):
                return {
                    "number": 900,
                    "merged": False,
                    "merged_at": None,
                    "state": "open",
                    "head": {"ref": "feat/x", "sha": "a" * 40},
                }
            raise AssertionError(f"unexpected {method} {url}")
        self.mock_api.side_effect = fake_api
        res = gitea_reconcile_merged_cleanups(
            dry_run=False,
            execute_confirmed=True,
            pr_number=900,
            remote="prgs",
        )
        self.assertFalse(res.get("success"))
        self.assertFalse(res.get("performed"))
        self.assertEqual(res.get("blocker_kind"), "pr_not_merged")
    def test_exact_pr_invalid_number_fails_closed(self):
        from mcp_server import gitea_reconcile_merged_cleanups
        res = gitea_reconcile_merged_cleanups(
            dry_run=True,
            pr_number=0,
            remote="prgs",
        )
        self.assertFalse(res.get("success"))
        self.assertEqual(res.get("blocker_kind"), "invalid_pr_number")
        self.mock_api.assert_not_called()
    def test_batch_mode_still_works_without_pr_number(self):
        """Unfiltered batch path remains backward compatible."""
        from mcp_server import gitea_reconcile_merged_cleanups
        self.mock_all.side_effect = lambda url, auth, limit=None: []
        self.mock_api.side_effect = lambda *a, **k: {}
        patch(
            "mcp_server.merged_cleanup_reconcile.build_reconciliation_report",
            return_value={
                "entries": [],
                "reviewer_scratch_entries": [],
                "merged_pr_count": 0,
            },
        ).start()
        res = gitea_reconcile_merged_cleanups(dry_run=True, remote="prgs", limit=10)
        self.assertTrue(res.get("success"))
        self.assertEqual(res.get("selection_mode"), "batch")
        self.assertIsNone(res.get("selected_pr_number"))
 if __name__ == "__main__":
    unittest.main()
@@ -0,0 +1,244 @@
 """#855 AC4: an expired reviewer lease must not indefinitely protect an
 already-merged branch when no live claimant exists.
 Two layers are covered:
 * ``branch_cleanup_guard.assess_expired_reviewer_lease_reclaim`` — the pure,
  fail-closed reclaim decision. Every condition must be provably satisfied or
  the lease keeps protecting the branch.
 * ``gitea_mcp_server._collect_branch_ownership_records`` — the wiring that
  supplies authoritative evidence (PR merged state, owner-process liveness,
  competing ownership) to that decision, and flips an expired reviewer lease
  to reclaimable only under the full policy.
 All inputs are fabricated; no real repository, lease, or credential is used.
 """
 import importlib
 import unittest
 from unittest.mock import patch
 import branch_cleanup_guard
 mcp_server = importlib.import_module("gitea_mcp_server")
 FAKE_AUTH = "token fake"
 REMOTE = "prgs"
 ORG = "Scaled-Tech-Consulting"
 REPO = "Gitea-Tools"
 HOST = "gitea.prgs.cc"
 BRANCH = "feat/issue-638-webui-app-shell-phase1"
 PR_NUMBER = 818
 class TestAssessExpiredReviewerLeaseReclaim(unittest.TestCase):
    """Pure fail-closed reclaim decision (#855 AC4)."""
    def _call(self, **overrides):
        base = dict(
            role="reviewer",
            status="expired",
            pr_merged=True,
            owner_pid_alive=False,
            competing_active_claimant=False,
        )
        base.update(overrides)
        return branch_cleanup_guard.assess_expired_reviewer_lease_reclaim(**base)
    def test_full_policy_satisfied_allows_reclaim(self):
        out = self._call()
        self.assertTrue(out["reclaim_allowed"])
        self.assertEqual(out["reasons"], [])
        self.assertEqual(out["decision"], "reclaim_expired_reviewer_lease")
    def test_stale_dead_process_reviewer_also_reclaimable(self):
        out = self._call(status="stale_dead_process")
        self.assertTrue(out["reclaim_allowed"])
    def test_non_reviewer_role_never_reclaims(self):
        for role in ("author", "merger", "controller", "reconciler", "unknown"):
            with self.subTest(role=role):
                out = self._call(role=role)
                self.assertFalse(out["reclaim_allowed"])
                self.assertTrue(out["reasons"])
                self.assertEqual(out["decision"], "keep_protecting")
    def test_active_status_never_reclaims(self):
        out = self._call(status="active")
        self.assertFalse(out["reclaim_allowed"])
    def test_pr_not_merged_blocks_reclaim(self):
        out = self._call(pr_merged=False)
        self.assertFalse(out["reclaim_allowed"])
    def test_pr_merged_unknown_fails_closed(self):
        out = self._call(pr_merged=None)
        self.assertFalse(out["reclaim_allowed"])
    def test_owner_process_alive_blocks_reclaim(self):
        out = self._call(owner_pid_alive=True)
        self.assertFalse(out["reclaim_allowed"])
    def test_owner_liveness_unknown_fails_closed(self):
        out = self._call(owner_pid_alive=None)
        self.assertFalse(out["reclaim_allowed"])
    def test_competing_active_claimant_blocks_reclaim(self):
        out = self._call(competing_active_claimant=True)
        self.assertFalse(out["reclaim_allowed"])
    def test_competing_claimant_unknown_fails_closed(self):
        out = self._call(competing_active_claimant=None)
        self.assertFalse(out["reclaim_allowed"])
    def test_reasons_never_leak_secrets(self):
        out = self._call(role="author")
        blob = " ".join(out["reasons"]).lower()
        self.assertNotIn("token", blob)
        self.assertNotIn("password", blob)
 class _FakeLease(dict):
    pass
 class TestCollectorExpiredReviewerReclaimWiring(unittest.TestCase):
    """`_collect_branch_ownership_records` supplies authoritative evidence and
    flips an expired reviewer lease to reclaimable only under the full policy."""
    def _run(
        self,
        *,
        lease_role="reviewer",
        lease_freshness="stale_dead_process",
        owner_pid_alive=False,
        pr_merged=True,
        extra_leases=None,
        worktree_on_branch=False,
    ):
        lease = _FakeLease(
            role=lease_role,
            work_kind="pr",
            work_number=PR_NUMBER,
            branch=BRANCH,
            status="active",
            owner_pid=999999,
            remote=REMOTE,
            org=ORG,
            repo=REPO,
            host=HOST,
            freshness={
                "freshness": lease_freshness,
                "owner_pid": 999999,
                "owner_pid_alive": owner_pid_alive,
                "expired_by_time": lease_freshness == "expired",
            },
        )
        leases = [lease] + list(extra_leases or [])
        pr_payload = {
            "number": PR_NUMBER,
            "merged": pr_merged,
            "merged_at": "2026-07-23T00:00:00Z" if pr_merged else None,
            "head": {"ref": BRANCH},
        }
        def fake_api_request(method, url, *a, **k):
            if method == "GET" and f"/pulls/{PR_NUMBER}" in url:
                return pr_payload
            raise AssertionError(f"unexpected api_request {method} {url}")
        wt_entries = []
        if worktree_on_branch:
            wt_entries = [{"branch": BRANCH, "path": f"/x/branches/{BRANCH}"}]
        with patch.object(
            mcp_server.lease_lifecycle,
            "list_active_leases",
            return_value={"leases": leases},
        ), patch.object(
            mcp_server.control_plane_db, "get_db", return_value=object(), create=True
        ), patch.object(
            mcp_server.issue_lock_store, "iter_lock_files", return_value=[]
        ), patch.object(
            mcp_server.worktree_cleanup_audit,
            "list_worktrees",
            return_value=wt_entries,
        ), patch.object(
            mcp_server, "api_get_all", return_value=[]
        ), patch.object(
            mcp_server, "api_request", side_effect=fake_api_request
        ):
            return mcp_server._collect_branch_ownership_records(
                remote=REMOTE,
                host=HOST,
                org=ORG,
                repo=REPO,
                branch=BRANCH,
                pr_number=PR_NUMBER,
                project_root="/x",
                auth=FAKE_AUTH,
                base_api="https://gitea.prgs.cc/api/v1/repos/x/y",
            )
    def _reviewer_records(self, bundle):
        return [
            rec
            for rec in bundle["records"]
            if rec.get("category")
            == branch_cleanup_guard.OWNERSHIP_CATEGORY_REVIEWER_LEASE
        ]
    def test_merged_dead_uncontested_reviewer_lease_is_reclaimable(self):
        bundle = self._run()
        self.assertFalse(bundle["inventory_error"])
        recs = self._reviewer_records(bundle)
        self.assertEqual(len(recs), 1)
        self.assertTrue(recs[0]["reclaim_allowed"])
        # And the guard consequently does not block deletion on it.
        ownership = branch_cleanup_guard.assess_active_branch_ownership(
            remote=REMOTE, org=ORG, repo=REPO, branch=BRANCH, host=HOST,
            records=bundle["records"],
        )
        self.assertFalse(ownership["block"])
    def test_unmerged_pr_keeps_reviewer_lease_protective(self):
        bundle = self._run(pr_merged=False)
        recs = self._reviewer_records(bundle)
        self.assertEqual(len(recs), 1)
        self.assertFalse(recs[0]["reclaim_allowed"])
        ownership = branch_cleanup_guard.assess_active_branch_ownership(
            remote=REMOTE, org=ORG, repo=REPO, branch=BRANCH, host=HOST,
            records=bundle["records"],
        )
        self.assertTrue(ownership["block"])
    def test_owner_process_alive_keeps_reviewer_lease_protective(self):
        bundle = self._run(owner_pid_alive=True, lease_freshness="expired")
        recs = self._reviewer_records(bundle)
        self.assertFalse(recs[0]["reclaim_allowed"])
    def test_competing_worktree_binding_keeps_reviewer_lease_protective(self):
        bundle = self._run(worktree_on_branch=True)
        recs = self._reviewer_records(bundle)
        self.assertFalse(recs[0]["reclaim_allowed"])
        ownership = branch_cleanup_guard.assess_active_branch_ownership(
            remote=REMOTE, org=ORG, repo=REPO, branch=BRANCH, host=HOST,
            records=bundle["records"],
        )
        self.assertTrue(ownership["block"])
    def test_expired_author_lease_never_reclaimed_by_reviewer_policy(self):
        bundle = self._run(lease_role="author")
        author_recs = [
            rec
            for rec in bundle["records"]
            if rec.get("category")
            == branch_cleanup_guard.OWNERSHIP_CATEGORY_AUTHOR_LEASE
        ]
        self.assertEqual(len(author_recs), 1)
        self.assertFalse(author_recs[0]["reclaim_allowed"])
 if __name__ == "__main__":
    unittest.main()
@@ -0,0 +1,630 @@
 """Durable linked-issue lock head refresh + merge-sync dead-session recovery (#871).
 ``gitea_update_pr_branch_by_merge`` advances a PR's *remote* head but historically
 never advanced the linked durable issue lock's recorded head. After the owning
 session died the drifted lock became unrecoverable and no further synchronization
 was possible (PR #866 / issue #855).
 Two halves are covered:
 * the write-side refresh (``issue_lock_store.assess/apply_durable_lock_head_refresh``)
  that records the new synced head under compare-and-swap with read-after-write; and
 * the read-side recovery relation (``issue_lock_recovery`` +
  ``issue_lock_worktree.read_merge_sync_provenance``) that lets a dead-session lock
  whose recorded head is a merge-sync *ancestor* of the live PR head be recovered —
  and nothing else.
 """
 from __future__ import annotations
 import os
 import subprocess
 import sys
 import tempfile
 import unittest
 from datetime import datetime, timedelta, timezone
 from pathlib import Path
 sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
 import issue_lock_recovery  # noqa: E402
 import issue_lock_store  # noqa: E402
 import issue_lock_worktree  # noqa: E402
 ISSUE = 8710
 PR_NUMBER = 8711
 BRANCH = f"fix/issue-{ISSUE}-durable-lock-head-refresh"
 IDENTITY = "example-user"
 PROFILE = "example-author"
 OLD = "a" * 40
 NEW1 = "b" * 40
 NEW2 = "c" * 40
 BASE = "d" * 40
 REMOTE = "prgs"
 ORG = "ExampleOrg"
 REPO = "ExampleRepo"
 def dead_pid() -> int:
    proc = subprocess.Popen([sys.executable, "-c", "pass"])
    proc.wait()
    return proc.pid
 def future_ts(hours: int = 4) -> str:
    return (
        (datetime.now(timezone.utc) + timedelta(hours=hours))
        .isoformat()
        .replace("+00:00", "Z")
    )
 def _git(cwd, *args):
    return subprocess.run(
        ["git", "-C", cwd, *args],
        capture_output=True,
        text=True,
        check=True,
    )
 def _rev(cwd, ref="HEAD") -> str:
    return _git(cwd, "rev-parse", ref).stdout.strip()
 def build_merge_sync_repo(tmp: str) -> dict:
    """Build a repo where a feature branch was synced by merging master in.
    Returns a dict with the prior (branch) head, the synced merge-commit head,
    the master tip, plus a rebase-style linear descendant and an unrelated head.
    """
    _git(tmp, "init", "-q", "-b", "master")
    _git(tmp, "config", "user.email", "[email protected]")
    _git(tmp, "config", "user.name", "T")
    Path(tmp, "base.txt").write_text("base\n")
    _git(tmp, "add", "-A")
    _git(tmp, "commit", "-q", "-m", "root")
    # Feature branch cut from root, one commit — this is the PRIOR/recorded head.
    _git(tmp, "checkout", "-q", "-b", BRANCH)
    Path(tmp, "feature.txt").write_text("feature\n")
    _git(tmp, "add", "-A")
    _git(tmp, "commit", "-q", "-m", "feature work")
    prior = _rev(tmp)
    # Master advances (the base the sync will merge in).
    _git(tmp, "checkout", "-q", "master")
    Path(tmp, "base.txt").write_text("base\nmore\n")
    _git(tmp, "add", "-A")
    _git(tmp, "commit", "-q", "-m", "master advance")
    master_tip = _rev(tmp)
    # Sync: merge master INTO the feature branch → merge commit, first parent = prior.
    _git(tmp, "checkout", "-q", BRANCH)
    _git(tmp, "merge", "-q", "--no-ff", "-m", "Merge master into feature", "master")
    synced = _rev(tmp)
    # A plain linear descendant of prior (NOT a merge) — a rebase/extra-commit shape.
    _git(tmp, "checkout", "-q", "-b", "linear-branch", prior)
    Path(tmp, "extra.txt").write_text("extra\n")
    _git(tmp, "add", "-A")
    _git(tmp, "commit", "-q", "-m", "extra linear commit")
    linear = _rev(tmp)
    # An unrelated root (force-push / rewritten history shape).
    unrelated_dir = tempfile.mkdtemp()
    _git(unrelated_dir, "init", "-q", "-b", "x")
    _git(unrelated_dir, "config", "user.email", "[email protected]")
    _git(unrelated_dir, "config", "user.name", "T")
    Path(unrelated_dir, "z.txt").write_text("z\n")
    _git(unrelated_dir, "add", "-A")
    _git(unrelated_dir, "commit", "-q", "-m", "unrelated")
    unrelated = _rev(unrelated_dir)
    # Leave the worktree checked out on the feature branch at the PRIOR head, as
    # a dead author session that never advanced would have left it.
    _git(tmp, "checkout", "-q", BRANCH)
    _git(tmp, "reset", "-q", "--hard", prior)
    return {
        "prior": prior,
        "master_tip": master_tip,
        "synced": synced,
        "linear": linear,
        "unrelated": unrelated,
    }
 # ─────────────────────────── write-side refresh ───────────────────────────
 class TestDurableLockHeadRefresh(unittest.TestCase):
    def setUp(self):
        self.lock_dir = tempfile.mkdtemp()
        self.wt = tempfile.mkdtemp()
        lock_data = {
            "issue_number": ISSUE,
            "branch_name": BRANCH,
            "worktree_path": self.wt,
            "remote": REMOTE,
            "org": ORG,
            "repo": REPO,
            "claimant": {"username": IDENTITY, "profile": PROFILE},
            "work_lease": {
                "operation_type": issue_lock_store.AUTHOR_ISSUE_WORK_LEASE,
                "issue_number": ISSUE,
                "branch": BRANCH,
                "worktree_path": self.wt,
                "claimant": {"username": IDENTITY, "profile": PROFILE},
                "expires_at": future_ts(),
            },
        }
        issue_lock_store.bind_session_lock(lock_data, lock_dir=self.lock_dir)
    def _apply(self, **over):
        kw = dict(
            remote=REMOTE, org=ORG, repo=REPO, issue_number=ISSUE,
            branch_name=BRANCH, worktree_path=self.wt, pr_number=PR_NUMBER,
            identity=IDENTITY, profile=PROFILE, current_pid=os.getpid(),
            expected_old_head=OLD, new_head=NEW1, synced_at=future_ts(0),
            base_head=BASE, lock_dir=self.lock_dir,
        )
        kw.update(over)
        return issue_lock_store.apply_durable_lock_head_refresh(**kw)
    def _load(self):
        return issue_lock_store.load_issue_lock(
            remote=REMOTE, org=ORG, repo=REPO, issue_number=ISSUE,
            lock_dir=self.lock_dir,
        )
    def test_first_sync_updates_recorded_head(self):
        """AC1: first base sync writes the resulting head to the durable lock."""
        res = self._apply()
        self.assertTrue(res["refreshed"], res["reasons"])
        self.assertTrue(res["read_after_write_ok"])
        self.assertEqual(self._load().get("synced_pr_head"), NEW1)
    def test_second_sync_after_master_advance(self):
        """AC2: a later master advance permits a second sanctioned sync."""
        self.assertTrue(self._apply()["refreshed"])
        res2 = self._apply(expected_old_head=NEW1, new_head=NEW2)
        self.assertTrue(res2["refreshed"], res2["reasons"])
        self.assertEqual(self._load().get("synced_pr_head"), NEW2)
        history = self._load().get("branch_sync_history")
        self.assertEqual(len(history), 2)
        self.assertEqual(history[0]["last_synced_pr_head"], NEW1)
        self.assertEqual(history[1]["prior_pr_head"], NEW1)
    def test_cas_detects_concurrent_head_change(self):
        """AC6: CAS refuses when the recorded synced head is not the old head."""
        self.assertTrue(self._apply()["refreshed"])  # recorded head now NEW1
        # A second sync claiming the old head is still OLD must fail closed.
        res = self._apply(expected_old_head=OLD, new_head=NEW2)
        self.assertFalse(res["refreshed"])
        self.assertTrue(any("CAS" in r or "concurrent" in r for r in res["reasons"]))
        self.assertEqual(self._load().get("synced_pr_head"), NEW1)
    def test_wrong_issue_fails_closed(self):
        res = self._apply(issue_number=999999)
        self.assertFalse(res["refreshed"])
    def test_wrong_branch_fails_closed(self):
        res = self._apply(branch_name="fix/issue-8710-wrong")
        self.assertFalse(res["refreshed"])
    def test_wrong_repo_fails_closed(self):
        res = self._apply(repo="OtherRepo")
        self.assertFalse(res["refreshed"])
    def test_wrong_identity_fails_closed(self):
        res = self._apply(identity="intruder")
        self.assertFalse(res["refreshed"])
    def test_wrong_profile_fails_closed(self):
        res = self._apply(profile="prgs-reviewer")
        self.assertFalse(res["refreshed"])
    def test_foreign_session_fails_closed(self):
        """A refresh is not a recovery: the current process must own the lock."""
        path = issue_lock_store.lock_file_path(
            remote=REMOTE, org=ORG, repo=REPO, issue_number=ISSUE,
            lock_dir=self.lock_dir,
        )
        rec = issue_lock_store.read_lock_file(path)
        rec["session_pid"] = dead_pid()
        rec["pid"] = rec["session_pid"]
        issue_lock_store.save_lock_file(path, rec)
        res = self._apply()
        self.assertFalse(res["refreshed"])
        self.assertTrue(any("current session" in r or "live owner" in r for r in res["reasons"]))
    def test_new_equals_old_fails_closed(self):
        res = self._apply(expected_old_head=OLD, new_head=OLD)
        self.assertFalse(res["refreshed"])
    def test_non_full_sha_fails_closed(self):
        self.assertFalse(self._apply(new_head="deadbeef")["refreshed"])
        self.assertFalse(self._apply(expected_old_head="xyz")["refreshed"])
    def test_no_lock_fails_closed(self):
        assessment = issue_lock_store.assess_durable_lock_head_refresh(
            None, remote=REMOTE, org=ORG, repo=REPO, issue_number=ISSUE,
            branch_name=BRANCH, worktree_path=self.wt, pr_number=PR_NUMBER,
            identity=IDENTITY, profile=PROFILE, current_pid=os.getpid(),
            expected_old_head=OLD, new_head=NEW1,
        )
        self.assertFalse(assessment["allowed"])
 # ─────────────────────── merge-sync provenance (real git) ───────────────────
 class TestMergeSyncProvenanceObservation(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.mkdtemp()
        self.shas = build_merge_sync_repo(self.tmp)
    def test_merge_sync_is_recognized(self):
        obs = issue_lock_worktree.read_merge_sync_provenance(
            self.tmp, prior_head_sha=self.shas["prior"],
            synced_head_sha=self.shas["synced"],
        )
        self.assertTrue(obs["is_merge_sync"], obs["reasons"])
        self.assertTrue(obs["prior_is_ancestor"])
        self.assertTrue(obs["synced_is_merge"])
        self.assertTrue(obs["first_parent_reaches_prior"])
    def test_linear_descendant_is_not_a_merge_sync(self):
        """A plain non-merge descendant (rebase/extra commit) is not a sync."""
        obs = issue_lock_worktree.read_merge_sync_provenance(
            self.tmp, prior_head_sha=self.shas["prior"],
            synced_head_sha=self.shas["linear"],
        )
        self.assertTrue(obs["probe_ok"])
        self.assertFalse(obs["is_merge_sync"])
        self.assertFalse(obs["synced_is_merge"])
    def test_unrelated_history_fails_closed(self):
        """A rewritten/force-pushed head where prior is unreachable fails closed."""
        obs = issue_lock_worktree.read_merge_sync_provenance(
            self.tmp, prior_head_sha=self.shas["prior"],
            synced_head_sha=self.shas["unrelated"],
        )
        self.assertFalse(obs["is_merge_sync"])
    def test_missing_args_fail_closed(self):
        obs = issue_lock_worktree.read_merge_sync_provenance(
            self.tmp, prior_head_sha=None, synced_head_sha=self.shas["synced"],
        )
        self.assertFalse(obs["is_merge_sync"])
 # ──────────────────── merge-sync dead-session recovery ──────────────────────
 def make_dead_lock(worktree, **over):
    pid = dead_pid()
    lock = {
        "issue_number": ISSUE,
        "branch_name": BRANCH,
        "worktree_path": worktree,
        "remote": REMOTE,
        "org": ORG,
        "repo": REPO,
        "session_pid": pid,
        "pid": pid,
        "claimant": {"username": IDENTITY, "profile": PROFILE},
        "work_lease": {
            "operation_type": issue_lock_store.AUTHOR_ISSUE_WORK_LEASE,
            "issue_number": ISSUE,
            "branch": BRANCH,
            "worktree_path": worktree,
            "claimant": {"username": IDENTITY, "profile": PROFILE},
            "expires_at": future_ts(),
        },
    }
    lock.update(over)
    return lock
 def sync_prov(prior, synced, **over):
    d = {
        "prior_head_sha": prior,
        "synced_head_sha": synced,
        "probe_ok": True,
        "prior_present": True,
        "synced_present": True,
        "prior_is_ancestor": True,
        "synced_is_merge": True,
        "first_parent_reaches_prior": True,
        "is_merge_sync": True,
        "first_parent_sha": prior,
        "parent_count": 2,
        "proof": f"{synced} merged base into branch above {prior}",
        "reasons": [],
    }
    d.update(over)
    return d
 class TestMergeSyncRecovery(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.mkdtemp()
        self.shas = build_merge_sync_repo(self.tmp)
        self.prior = self.shas["prior"]
        self.synced = self.shas["synced"]
    def _assess(self, **over):
        lock = over.pop("_lock", None) or make_dead_lock(self.tmp)
        kw = dict(
            issue_number=ISSUE, branch_name=BRANCH, worktree_path=self.tmp,
            remote=REMOTE, org=ORG, repo=REPO, identity=IDENTITY, profile=PROFILE,
            current_branch=BRANCH, porcelain_status="",
            head_sha=self.prior, remote_head_sha=self.synced,
            pr_head_sha=self.synced, pr_number=PR_NUMBER,
            competing_live_locks=[], candidate_branches=[BRANCH],
            current_pid=os.getpid(),
            remote_branch_exists=True,
            sync_provenance=sync_prov(self.prior, self.synced),
        )
        kw.update(over)
        return issue_lock_recovery.assess_dead_session_lock_recovery(lock, **kw)
    def test_merge_sync_drift_is_recoverable(self):
        """AC3/AC4: dead session, recorded head is a merge-sync ancestor of PR head."""
        res = self._assess()
        self.assertEqual(res["outcome"], issue_lock_recovery.RECOVERY_SANCTIONED, res["reasons"])
        self.assertEqual(
            res["evidence"]["head_relation"],
            issue_lock_recovery.HEAD_RELATION_REMOTE_MERGE_SYNCED,
        )
        self.assertEqual(res["evidence"]["accepted_head"], self.synced)
    def test_missing_provenance_fails_closed(self):
        """No server-derived provenance → cannot accept a remote ahead of local."""
        res = self._assess(sync_provenance=None)
        self.assertEqual(res["outcome"], issue_lock_recovery.REFUSED)
    def test_non_ancestor_recorded_head_fails_closed(self):
        """AC7: provenance that does not prove ancestry is rejected."""
        res = self._assess(
            sync_provenance=sync_prov(
                self.prior, self.synced, prior_is_ancestor=False, is_merge_sync=False,
                reasons=["prior head is not an ancestor"],
            )
        )
        self.assertEqual(res["outcome"], issue_lock_recovery.REFUSED)
    def test_force_pushed_history_fails_closed(self):
        """AC8: a rewritten head (not a merge sync) stays protected."""
        res = self._assess(
            sync_provenance=sync_prov(
                self.prior, self.synced, is_merge_sync=False, synced_is_merge=False,
                reasons=["not a merge-based sync"],
            )
        )
        self.assertEqual(res["outcome"], issue_lock_recovery.REFUSED)
    def test_provenance_for_other_commits_fails_closed(self):
        """Provenance whose endpoints differ from the heads under assessment is rejected."""
        res = self._assess(
            sync_provenance=sync_prov("f" * 40, self.synced),
        )
        self.assertEqual(res["outcome"], issue_lock_recovery.REFUSED)
    def test_dirty_worktree_fails_closed(self):
        """AC11: dirty worktrees remain protected."""
        res = self._assess(porcelain_status=" M feature.txt\n")
        self.assertEqual(res["outcome"], issue_lock_recovery.REFUSED)
    def test_live_owner_fails_closed(self):
        """AC10: a live recorded owner is not a dead-session recovery."""
        lock = make_dead_lock(self.tmp, session_pid=os.getpid(), pid=os.getpid())
        res = self._assess(_lock=lock)
        self.assertEqual(res["outcome"], issue_lock_recovery.REFUSED)
    def test_competing_claimant_fails_closed(self):
        """AC13: a competing live lock blocks recovery."""
        res = self._assess(
            competing_live_locks=[{
                "issue_number": ISSUE, "branch_name": BRANCH,
                "worktree_path": "/some/other/wt", "pid": os.getpid(),
            }]
        )
        self.assertEqual(res["outcome"], issue_lock_recovery.REFUSED)
    def test_wrong_branch_fails_closed(self):
        """AC9: worktree on a different branch fails closed."""
        res = self._assess(current_branch="fix/issue-8710-other")
        self.assertEqual(res["outcome"], issue_lock_recovery.REFUSED)
    def test_wrong_identity_fails_closed(self):
        res = self._assess(identity="intruder")
        self.assertEqual(res["outcome"], issue_lock_recovery.REFUSED)
    def test_pr_head_mismatch_fails_closed(self):
        """The open PR must sit at the synced remote head."""
        res = self._assess(pr_head_sha="e" * 40)
        self.assertEqual(res["outcome"], issue_lock_recovery.REFUSED)
    def test_owning_pr_evidence_for_merge_sync(self):
        res = self._assess()
        ev = issue_lock_recovery.owning_pr_recovery_evidence(res)
        self.assertIsNotNone(ev)
        self.assertEqual(ev["pr_number"], PR_NUMBER)
        self.assertEqual(ev["head_sha"], self.synced)
        self.assertEqual(
            ev["head_relation"],
            issue_lock_recovery.HEAD_RELATION_REMOTE_MERGE_SYNCED,
        )
    def test_recovered_owning_pr_from_persisted_record(self):
        res = self._assess()
        record = issue_lock_recovery.build_recovery_record(res, recovered_at=future_ts(0))
        lock = {"issue_number": ISSUE, "branch_name": BRANCH,
                "dead_session_recovery": record}
        rebuilt = issue_lock_recovery.recovered_owning_pr_from_lock(lock)
        self.assertIsNotNone(rebuilt)
        self.assertEqual(rebuilt["head_sha"], self.synced)
        self.assertEqual(
            rebuilt["head_relation"],
            issue_lock_recovery.HEAD_RELATION_REMOTE_MERGE_SYNCED,
        )
 class TestExistingRelationsUnchanged(unittest.TestCase):
    """AC14/AC15: equal-head recovery still works; merge-sync did not weaken it."""
    def setUp(self):
        self.tmp = tempfile.mkdtemp()
        self.shas = build_merge_sync_repo(self.tmp)
    def test_equal_head_recovery_still_sanctioned(self):
        # Worktree at prior head; remote also at prior head → the #753 equal case.
        prior = self.shas["prior"]
        lock = make_dead_lock(self.tmp)
        res = issue_lock_recovery.assess_dead_session_lock_recovery(
            lock, issue_number=ISSUE, branch_name=BRANCH, worktree_path=self.tmp,
            remote=REMOTE, org=ORG, repo=REPO, identity=IDENTITY, profile=PROFILE,
            current_branch=BRANCH, porcelain_status="",
            head_sha=prior, remote_head_sha=prior,
            pr_head_sha=prior, pr_number=PR_NUMBER,
            competing_live_locks=[], candidate_branches=[BRANCH],
            current_pid=os.getpid(), remote_branch_exists=True,
        )
        self.assertEqual(res["outcome"], issue_lock_recovery.RECOVERY_SANCTIONED, res["reasons"])
        self.assertEqual(
            res["evidence"]["head_relation"], issue_lock_recovery.HEAD_RELATION_EQUAL,
        )
 class TestUpdatePrWrapperPartialFailure(unittest.TestCase):
    """AC5/AC16: the tool advances the remote head then refreshes the durable lock.
    When the durable refresh fails after the remote advance, the tool must report a
    partial lifecycle failure and NOT a fully successful synchronization. Exact PR-
    head / base-head pinning is preserved (delegated to the real preflight, stubbed
    here only to isolate the post-update lifecycle branch).
    """
    def setUp(self):
        import gitea_mcp_server as gms  # noqa: E402
        self.gms = gms
        self._orig = {}
        def _patch(name, value):
            self._orig[name] = getattr(gms, name)
            setattr(gms, name, value)
        _patch("get_profile", lambda *a, **k: {
            "allowed_operations": ["gitea.branch.push"],
            "forbidden_operations": [],
            "profile_name": "prgs-author",
        })
        _patch("_role_kind", lambda *a, **k: "author")
        _patch("_profile_operation_gate", lambda *a, **k: None)
        _patch("_permission_block_report", lambda *a, **k: {})
        _patch("_resolve", lambda *a, **k: ("gitea.prgs.cc", ORG, REPO))
        _patch("_verify_role_mutation_workspace", lambda *a, **k: None)
        _patch("_get_workspace_porcelain", lambda *a, **k: "")
        _patch("_canonical_local_git_root", lambda *a, **k: "/x")
        _patch("_master_parity_block", lambda *a, **k: None)
        _patch("_auth", lambda *a, **k: {"token": "x"})
        _patch("repo_api_url", lambda *a, **k: "http://api")
        _patch("_redact", lambda s: s)
        _patch("_work_lease_claimant", lambda *a, **k: {
            "username": IDENTITY, "profile": PROFILE,
        })
        _patch("_prove_author_ownership_for_pr", lambda *a, **k: {
            "has_author_lock": True, "matched_issue": ISSUE,
            "matched_via": "branch", "linked_issues": [ISSUE],
            "recovered_owning_pr": None, "reasons": [],
        })
        # Real preflight is unit-tested elsewhere; stub it to isolate the
        # post-update durable-lock lifecycle branch under test.
        orig_pf = gms.pr_sync_status.assess_update_pr_branch_preflight
        self._orig_pf = orig_pf
        gms.pr_sync_status.assess_update_pr_branch_preflight = (
            lambda *a, **k: {"mutation_allowed": True, "reasons": [], "performed": False}
        )
        # Sequence the two GET /pulls calls: OLD before update, NEW after.
        self._pull_calls = {"n": 0}
        def fake_api_request(method, url, auth, *a, **k):
            m = method.upper()
            if m == "GET" and url.endswith(f"/pulls/{PR_NUMBER}"):
                self._pull_calls["n"] += 1
                head = OLD if self._pull_calls["n"] == 1 else NEW1
                return {
                    "state": "open",
                    "head": {"sha": head, "ref": BRANCH},
                    "base": {"sha": BASE, "ref": "master"},
                    "mergeable": True, "title": "t", "body": "b",
                }
            if m == "GET" and "/branches/" in url:
                return {"commit": {"id": BASE}}
            if m == "POST" and "/update" in url:
                return {}
            return {}
        _patch("api_request", fake_api_request)
    def tearDown(self):
        for name, value in self._orig.items():
            setattr(self.gms, name, value)
        self.gms.pr_sync_status.assess_update_pr_branch_preflight = self._orig_pf
    def _run(self):
        return self.gms.gitea_update_pr_branch_by_merge(
            pr_number=PR_NUMBER,
            expected_pr_head_sha=OLD,
            expected_base_head_sha=BASE,
            remote=REMOTE,
            worktree_path="/tmp/branches/wt-871",
        )
    def test_partial_failure_when_refresh_fails(self):
        self._orig["apply_durable_lock_head_refresh"] = (
            self.gms.issue_lock_store.apply_durable_lock_head_refresh
        )
        self.gms.issue_lock_store.apply_durable_lock_head_refresh = (
            lambda **k: {"refreshed": False, "reasons": ["forced refresh failure"]}
        )
        try:
            res = self._run()
        finally:
            self.gms.issue_lock_store.apply_durable_lock_head_refresh = (
                self._orig["apply_durable_lock_head_refresh"]
            )
        self.assertTrue(res["performed"])
        self.assertEqual(res["new_pr_head_sha"], NEW1)
        self.assertFalse(res["success"])
        self.assertTrue(res["partial_lifecycle_failure"])
        self.assertFalse(res["durable_lock_refreshed"])
    def test_full_success_when_refresh_succeeds(self):
        self._orig["apply_durable_lock_head_refresh"] = (
            self.gms.issue_lock_store.apply_durable_lock_head_refresh
        )
        self.gms.issue_lock_store.apply_durable_lock_head_refresh = (
            lambda **k: {"refreshed": True, "read_after_write_ok": True,
                         "new_head": NEW1, "reasons": ["ok"]}
        )
        try:
            res = self._run()
        finally:
            self.gms.issue_lock_store.apply_durable_lock_head_refresh = (
                self._orig["apply_durable_lock_head_refresh"]
            )
        self.assertTrue(res["success"])
        self.assertTrue(res["performed"])
        self.assertTrue(res["durable_lock_refreshed"])
        self.assertTrue(res["fully_synchronized"])
        self.assertEqual(res["new_pr_head_sha"], NEW1)
 if __name__ == "__main__":
    unittest.main()
@@ -0,0 +1,107 @@
 """Documentation acceptance for the MCP restart governance ADR (#656).
 Enforces issue #656 acceptance criteria:
 * AC1 — policy document exists with an authorization matrix and the recorded
  v1 decision (controller approval + automated safety gates).
 * AC2 — restart is stated as a last resort with enumerated narrower recoveries.
 * AC3 — a unilateral LLM full restart with affected sessions is forbidden.
 * AC4 — break-glass conditions are listed.
 * AC5 — the ADR is linked to #655, #652, #653, #630, #642, and is cross-linked
  from the safety model and the web-console deployment boundary docs.
 """
 from pathlib import Path
 REPO_ROOT = Path(__file__).resolve().parent.parent
 ADR = REPO_ROOT / "docs" / "architecture" / "mcp-restart-governance.md"
 ADR_BASENAME = "mcp-restart-governance.md"
 CROSS_LINK_DOCS = (
    REPO_ROOT / "docs" / "safety-model.md",
    REPO_ROOT / "docs" / "webui-deployment.md",
 )
 LINKED_ISSUES = ("#655", "#652", "#653", "#630", "#642")
 POLICY_IDS = ("RG-01", "RG-02", "RG-03", "RG-04", "RG-05", "RG-06", "RG-07", "RG-08")
 def _read(path: Path) -> str:
    assert path.is_file(), f"missing {path.relative_to(REPO_ROOT)}"
    return path.read_text(encoding="utf-8")
 def test_ac1_adr_exists_with_matrix_and_v1_decision():
    text = _read(ADR)
    lower = text.lower()
    assert text.lstrip().startswith("#"), "ADR lacks a title"
    assert "#656" in text
    assert "authorization matrix" in lower
    # The matrix is a real table with the worker and privileged roles.
    for role in ("author", "reviewer", "merger", "reconciler", "controller",
                 "operator", "admin"):
        assert role in lower, f"authorization matrix missing role {role!r}"
    # Recorded v1 decision.
    assert "restart-governance/v1" in text
    assert "controller approval" in lower and "automated safety gates" in lower
 def test_ac2_restart_is_last_resort_with_narrower_recoveries():
    text = _read(ADR)
    lower = text.lower()
    assert "last resort" in lower
    # Enumerated narrower recoveries precede full restart on the ladder.
    for rung in ("reconnect", "rebind", "scoped restart", "full restart",
                 "host"):
        assert rung in lower, f"recovery ladder missing rung {rung!r}"
 def test_ac3_forbids_unilateral_llm_full_restart_with_affected_sessions():
    text = _read(ADR)
    lower = text.lower()
    assert "forbidden" in lower
    assert "llm" in lower and "restart" in lower
    assert "unilateral" in lower
    # A worker role must not perform or authorize full/host restart.
    assert "must not" in lower
 def test_ac4_break_glass_conditions_listed():
    text = _read(ADR)
    lower = text.lower()
    assert "break-glass" in lower
    assert "incident" in lower
    assert "audit" in lower
 def test_ac5_adr_links_issue_lineage():
    text = _read(ADR)
    for issue in LINKED_ISSUES:
        assert issue in text, f"ADR must link issue {issue}"
 def test_ac5_safety_model_and_deployment_cross_link_adr():
    for path in CROSS_LINK_DOCS:
        text = _read(path)
        assert ADR_BASENAME in text, (
            f"{path.relative_to(REPO_ROOT)} must cross-link {ADR_BASENAME} "
            f"(issue #656 acceptance criterion 5)"
        )
 def test_policy_ids_present_for_enforcement_code():
    text = _read(ADR)
    for pid in POLICY_IDS:
        assert pid in text, f"policy id {pid} missing from ADR"
 def test_failure_behavior_denies_on_ambiguity():
    text = _read(ADR)
    lower = text.lower()
    assert "ambiguous" in lower and "deny" in lower
 def test_cross_links_do_not_embed_secrets():
    for path in (ADR,) + CROSS_LINK_DOCS:
        text = _read(path)
        for marker in ("ghp_", "BEGIN PRIVATE KEY", "Authorization: Bearer"):
            assert marker not in text, f"{path} contains {marker!r}"
@@ -139,8 +139,6 @@ 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",
Author	SHA1	Message	Date
jcwalker3	73f82a2305	Merge branch 'master' into fix/issue-855-pr-scoped-merged-cleanup	2026-07-24 03:15:43 -05:00
sysadmin	cc7dc8ac14	Merge pull request 'fix(author): refresh durable issue-lock head after branch sync; recover merge-sync-drifted dead-session locks (Closes #871 )' (#874 ) from fix/issue-871-durable-lock-head-refresh into master	2026-07-24 03:04:24 -05:00
jcwalker3	e151759212	Merge branch 'master' into fix/issue-871-durable-lock-head-refresh	2026-07-24 02:42:52 -05:00
sysadmin	60df5087e9	Merge pull request 'docs(governance): MCP restart governance and authorization policy (Closes #656 )' (#857 ) from docs/issue-656-mcp-restart-governance into master	2026-07-24 02:26:02 -05:00
jcwalker3	c9aa09f341	Merge branch 'master' into fix/issue-855-pr-scoped-merged-cleanup	2026-07-24 02:07:33 -05:00
jcwalker3	0088ecaf00	Merge branch 'master' into docs/issue-656-mcp-restart-governance	2026-07-24 02:07:17 -05:00
sysadminandClaude Opus 4.8	5d59c57c98	fix(author): refresh durable issue-lock head after branch sync; recover merge-sync-drifted dead-session locks (#871 ) `gitea_update_pr_branch_by_merge` advanced a PR's remote head but never advanced the linked durable issue lock's recorded head. After the owning session died the drifted lock became unrecoverable and no further synchronization was possible (PR #866 / issue #855). Write-side (prevents future drift): - issue_lock_store.assess/apply_durable_lock_head_refresh: on a successful sync, CAS-refresh the durable lock's recorded head from the exact expected PR head to the resulting head, re-verifying repo/issue/branch/worktree/ identity/profile/live-session ownership, with read-after-write verification. - gitea_update_pr_branch_by_merge now refreshes the lock after the remote advance and reports a PARTIAL LIFECYCLE FAILURE (success=False) when the refresh fails, instead of falsely reporting a full synchronization. Read-side (recovers already-drifted locks): - issue_lock_worktree.read_merge_sync_provenance: server-side git observation proving a remote head is a sanctioned base-into-branch merge that preserved the branch mainline back to the recorded head. - issue_lock_recovery: new HEAD_RELATION_REMOTE_MERGE_SYNCED accepts a dead-session lock whose recorded head is a strict merge-sync ancestor of the live PR head — and only that. Rewrites, rebases, force-pushes, non-ancestor heads, dirty worktrees, live/competing owners, and wrong repo/issue/branch/ identity/profile all stay protected. No existing exact-head, branch-protection, parity, workspace, identity, role, or mutation-safety gate is weakened. All provenance is server-derived; nothing is reachable from an MCP caller. Tests: tests/test_issue_871_durable_lock_head_refresh.py (32 cases) covering first/second sync, CAS, ownership, partial-failure, merge-sync recovery happy-path and every fail-closed branch. Full suite: 4789 passed, 13 pre- existing baseline failures unchanged. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]> Claude-Session: https://claude.ai/code/session_01FZPyVh2DGczQrDxtqwGH5p	2026-07-24 02:56:57 -04:00
jcwalker3	499b87c482	Merge branch 'master' into fix/issue-855-pr-scoped-merged-cleanup	2026-07-24 00:25:40 -05:00
sysadminandClaude Opus 4.8	24c52abf6b	feat(reconciler): PR-scoped post-merge cleanup executor + expired reviewer-lease reclaim (Closes #855 ) Adds a single-target path for post-merge cleanup so a reconciler can complete one merged PR without a batch sweep across unrelated PRs. ## PR-scoped selector gitea_reconcile_merged_cleanups gains an optional pr_number. When set, only that merged PR is assessed and acted on: the PR is resolved live and fails closed on an invalid/non-positive number, an unresolvable or ambiguous PR, or an unmerged PR; reviewer scratch worktrees are filtered to that PR; and the report's entry set is pinned to exactly [pr_number], failing closed on any drift. The existing execute loop then operates on the single pinned entry only -- worktree removal, ownership reassessment, then remote-branch delete -- with no unrelated target. Batch behaviour is unchanged when pr_number is omitted. ## Expired reviewer-lease reclaim (AC4) An expired or stale reviewer lease no longer protects an already-merged branch forever. branch_cleanup_guard.assess_expired_reviewer_lease_reclaim makes the decision explicitly and fail-closed: reclaim only when the lease is a reviewer lease, its status is expired/stale, the PR is proven merged, the owner process is proven dead, and no competing active claimant uses the branch. _collect_branch_ownership_records supplies that evidence from authoritative state (live PR merged-state, lease owner liveness, and the full ownership inventory for competing-claimant detection) and evaluates it only after the complete inventory is built, so the post-worktree-removal reassessment is what unblocks the branch delete. Any unknown fails closed. Author/merger/controller/reconciler leases are untouched; active leases, worktree bindings, issue locks, and live sessions still block. ## Tests - tests/test_issue_855_expired_reviewer_reclaim.py: full fail-closed matrix for the reclaim decision plus collector wiring (merged+dead+uncontested reclaims; unmerged, live-owner, competing-worktree, and author-lease cases stay protective). - tests/test_branch_cleanup_guard.py: exact-PR selector coverage (ignores newer PRs in the batch queue, execute mutates only the selected PR, unknown/not-merged/invalid fail closed, batch mode preserved). Changed-surface suites pass; the 2 pre-existing test_branch_cleanup_guard failures and test_reconciler_supersession_close reproduce identically on master `6d0015ca` and are unrelated to this change. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>	2026-07-24 00:49:52 -04:00
jcwalker3	dc99c15ffa	Merge branch 'master' into docs/issue-656-mcp-restart-governance	2026-07-23 19:53:16 -05:00
sysadminandClaude Opus 4.8	9f759150b8	docs(governance): correct linked-issue descriptions in restart ADR (#656 ) The Related section and cross-reference lines described #652, #653, #630, #642, and #591 by roles they do not hold. Align each description with the linked issue's actual title and scope: - #652 is the Control Plane Web Console product vision (restart controls live in its capability area A), not a restart-specific vision. - #653 is the console phased-delivery roadmap; restart controls are Phase 2. - #630 is the manual process-kill contamination guard, not the coordinator; the coordinator remains an unimplemented later child of #655. - #642 is the sanctioned restart / graceful reload console UX. - #591 is auto-restart on master advance (closed); only #584 is transport-flap reconnect. They were previously collapsed into one transport-recovery label. Documentation-only wording change. Policy IDs RG-01..RG-08, the policy version restart-governance/v1, the authorization matrix, and every normative statement are unchanged. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>	2026-07-23 19:18:47 -04:00
sysadmin	347464a057	Merge branch 'master' into docs/issue-656-mcp-restart-governance	2026-07-23 19:17:05 -04:00
sysadminandClaude Opus 4.8	1301a57de4	docs(governance): MCP restart governance and authorization policy (#656 ) Adds docs/architecture/mcp-restart-governance.md, the restart-governance/v1 ADR defining who may restart the MCP control plane and under what conditions. - Recovery ladder (reconnect -> rebind -> scoped restart -> full restart -> host) with restart stated as the last resort. - Authorization matrix across author/reviewer/merger/reconciler/controller/ operator/admin; no LLM worker role may perform or authorize a full or host restart. - v1 authority decision recorded: controller approval + automated safety gates; quorum deferred to a superseding ADR. - Break-glass path with pre-declared incident and mandatory post-hoc audit. - Ambiguous policy state denies restart. - Stable policy IDs RG-01..RG-08 for later enforcement code to bind to. Cross-links the ADR from docs/safety-model.md and docs/webui-deployment.md, and adds tests/test_mcp_restart_governance_docs.py asserting acceptance criteria 1-5. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>	2026-07-23 19:16:28 -04:00