Merge branch 'master' into feat/issue-727-pr-sync-status

Resolve task_capability_map conflict by keeping PR-sync entries and #725
merger-lease map entries. Fix author ownership for PR #728 (issue #727):
prove lock via linked issue/session/branch instead of issue_number==pr_number.
Add regression tests for 727/728 ownership and unrelated issue rejection.

.env.example

@@ -39,6 +39,27 @@ GITEA_AUDIT_LOG=/path/to/gitea-mcp-audit.log
 # only — never the token value. Surfaced by gitea_get_profile.
 GITEA_TOKEN_SOURCE=GITEA_TOKEN
 
+# ── Optional self-hosted Sentry observability (#606) ────────────────────────
+# Emits runtime errors, fail-closed workflow blockers, lease/terminal-lock/
+# stale-runtime collisions, and watchdog cron check-ins to a SELF-HOSTED Sentry
+# (https://sentry.prgs.cc/) — never Sentry Cloud. Gitea stays the source of
+# truth; Sentry is observe-only. OFF by default: with MCP_SENTRY_ENABLED unset
+# or SENTRY_DSN empty, nothing is initialised and no events are sent.
+#
+# Master gate. Truthy = 1/true/yes/on. Both this AND SENTRY_DSN are required.
+MCP_SENTRY_ENABLED=0
+# DSN for the self-hosted project (create a `gitea-tools-mcp` project in
+# https://sentry.prgs.cc/ and copy its DSN). Never commit a real DSN.
+SENTRY_DSN=
+# Deployment environment tag (local/dev/prod). Defaults to "development".
+SENTRY_ENVIRONMENT=development
+# Optional release identifier (e.g. a git SHA or version string).
+SENTRY_RELEASE=
+# Performance-trace sample rate, 0.0–1.0 (clamped). Default 0.0 (traces off).
+MCP_SENTRY_TRACES_SAMPLE_RATE=0.0
+# Set to 1 to forward Python logs to Sentry as structured logs. Default off.
+MCP_SENTRY_ENABLE_LOGS=0
+
 # Optional canonical runtime-profile config (#19). Instead of the fields above,
 # point every LLM launcher at ONE JSON file of named profiles and select one.
 # Secrets are referenced (keychain id / env var name), never inlined. See

anti_stomp_preflight.py

@@ -0,0 +1,807 @@
+"""Common anti-stomp preflight for every MCP mutation tool (#604).
+
+Even with leases, mutation tools need a **shared** preflight that prevents
+stale sessions, wrong worktrees, wrong repos, old prompts, terminal locks,
+foreign leases, contaminated approvals, and root-checkout mutations from
+slipping through.
+
+This module is the pure assessment core. Callers (MCP mutation entrypoints)
+gather live facts and pass them in — nothing here performs git, network, or
+durable-state I/O. Existing happy-path guards remain authoritative; this
+module composes them into one typed, fail-closed result.
+
+Required checks (issue #604):
+
+* repo/org verification
+* profile/role verification
+* root checkout clean and not used for mutation (when role requires branches/)
+* worktree under branches when required
+* active lease ownership (when lease is required for the mutation)
+* terminal lock status (when applicable)
+* expected head SHA (when pinned)
+* stale runtime status
+* workflow hash (when a workflow load is required)
+* source contamination status
+* no manual state/mtime/source-file bypass
+
+Failure returns a typed blocker and an exact next action. There is **no**
+agent-facing bypass flag.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+import author_mutation_worktree
+import master_parity_gate
+import remote_repo_guard
+import root_checkout_guard
+import stable_branch_push_guard
+
+# ── public constants ──────────────────────────────────────────────────────────
+
+# Typed blocker kinds returned in the structured result.
+BLOCKER_WRONG_REPO = "wrong_repo"
+BLOCKER_WRONG_ROLE = "wrong_role"
+BLOCKER_ROOT_CHECKOUT = "root_checkout_mutation"
+BLOCKER_WRONG_WORKTREE = "wrong_worktree"
+BLOCKER_FOREIGN_LEASE = "foreign_lease"
+BLOCKER_TERMINAL_LOCK = "terminal_lock"
+BLOCKER_HEAD_SHA = "head_sha_mismatch"
+BLOCKER_STALE_RUNTIME = "stale_runtime"
+BLOCKER_WORKFLOW_HASH = "workflow_hash"
+BLOCKER_SOURCE_CONTAMINATION = "source_contamination"
+BLOCKER_MANUAL_BYPASS = "manual_bypass"
+
+BLOCKER_KINDS = frozenset({
+    BLOCKER_WRONG_REPO,
+    BLOCKER_WRONG_ROLE,
+    BLOCKER_ROOT_CHECKOUT,
+    BLOCKER_WRONG_WORKTREE,
+    BLOCKER_FOREIGN_LEASE,
+    BLOCKER_TERMINAL_LOCK,
+    BLOCKER_HEAD_SHA,
+    BLOCKER_STALE_RUNTIME,
+    BLOCKER_WORKFLOW_HASH,
+    BLOCKER_SOURCE_CONTAMINATION,
+    BLOCKER_MANUAL_BYPASS,
+})
+
+# Mutation tasks that must invoke the shared anti-stomp preflight before
+# acting (issue #604 AC1). Every member must appear as a live task= kwarg
+# to verify_preflight_purity / _run_anti_stomp_preflight (or the review
+# anti_task map) in gitea_mcp_server.py. Inventory↔wiring tests fail if
+# a declared task is not wired.
+MUTATION_TASKS = frozenset({
+    "create_issue",
+    "comment_issue",
+    "close_issue",
+    "mark_issue",
+    "lock_issue",
+    "set_issue_labels",
+    "create_label",
+    "create_pr",
+    "close_pr",
+    "edit_pr",
+    "commit_files",
+    "gitea_commit_files",
+    "delete_branch",
+    "cleanup_merged_pr_branch",
+    "cleanup_stale_claims",
+    "reconcile_merged_cleanups",
+    "reconcile_already_landed_pr",
+    "reconcile_close_superseded_pr",
+    "post_heartbeat",
+    "acquire_reviewer_pr_lease",
+    "gitea_acquire_reviewer_pr_lease",
+    "adopt_merger_pr_lease",
+    "review_pr",
+    "submit_pr_review",
+    "approve_pr",
+    "request_changes_pr",
+    "merge_pr",
+})
+
+# Intentionally excluded from MUTATION_TASKS. Each has a dedicated
+# fail-closed gate that preserves decision-lock ownership, workflow-hash,
+# head-SHA, and repository consistency. Do not re-add without either
+# wiring shared preflight or updating this rationale (#604 Blocker B).
+DEDICATED_GATE_MUTATIONS: dict[str, str] = {
+    "mark_final_review_decision": (
+        "Local review-decision lock only. Enforced by session lock ownership, "
+        "workflow-hash, expected head SHA, PR work lease, eligibility, and "
+        "terminal-lock gates. No Gitea review POST; shared anti-stomp would "
+        "duplicate without side-effect-ordering value."
+    ),
+    "save_review_draft": (
+        "Local session-state draft save with live PR head/base consistency "
+        "checks. Not a Gitea review/merge mutation; dedicated head-SHA and "
+        "worktree resolution gates apply."
+    ),
+    "resume_review_draft": (
+        "Live-state resume with terminal lock, lease ownership, head/base SHA, "
+        "and parity checks. When submit=True, delegates to gitea_submit_pr_review "
+        "which runs shared anti-stomp before the Gitea review POST."
+    ),
+    "cleanup_stale_review_decision_lock": (
+        "Moot-lock cleanup with identity match, live PR merged/closed proof, "
+        "and reviewer capability gate (#594). Distinct from shared anti-stomp."
+    ),
+    "gitea_cleanup_stale_review_decision_lock": (
+        "Alias of cleanup_stale_review_decision_lock; dedicated #594 path."
+    ),
+    "heartbeat_reviewer_pr_lease": (
+        "Lease heartbeat posts via verify_preflight_purity(task='review_pr') "
+        "plus in-session lease ownership checks; not a separate mutation class."
+    ),
+    "release_reviewer_pr_lease": (
+        "Lease release uses workspace binding + ownership checks; posts only "
+        "when the session owns the active lease."
+    ),
+}
+
+# Roles that must mutate from a branches/ worktree (not the control checkout).
+_BRANCHES_REQUIRED_ROLES = frozenset({"author"})
+
+# Reviewer/merger mutations that require an owned lease + head pinning when
+# the caller supplies those facts.
+_LEASE_AWARE_TASKS = frozenset({
+    "review_pr",
+    "submit_pr_review",
+    "approve_pr",
+    "request_changes_pr",
+    "merge_pr",
+    "acquire_reviewer_pr_lease",
+    "gitea_acquire_reviewer_pr_lease",
+    "adopt_merger_pr_lease",
+})
+
+_NEXT_ACTIONS: dict[str, str] = {
+    BLOCKER_WRONG_REPO: (
+        "Pass explicit org= and repo= matching the local git remote "
+        "(e.g. org=Scaled-Tech-Consulting repo=Gitea-Tools) and retry."
+    ),
+    BLOCKER_WRONG_ROLE: (
+        "Call gitea_resolve_task_capability, switch to the required "
+        "namespace/profile, re-verify with gitea_whoami, then retry."
+    ),
+    BLOCKER_ROOT_CHECKOUT: (
+        "Leave the root control checkout on clean master; create or switch "
+        "to a session-owned worktree under branches/ and retry the mutation "
+        "with worktree_path set."
+    ),
+    BLOCKER_WRONG_WORKTREE: (
+        "Create or bind a session-owned worktree under branches/, set "
+        "worktree_path (or GITEA_ACTIVE_WORKTREE), and retry."
+    ),
+    BLOCKER_FOREIGN_LEASE: (
+        "Stop. Do not stomp a foreign lease. Wait for expiry, request "
+        "takeover through the sanctioned path, or hand off to the owner."
+    ),
+    BLOCKER_TERMINAL_LOCK: (
+        "Stop. A terminal review-decision lock is active for this head. "
+        "Do not re-approve/merge; follow the #332/#620 recovery path."
+    ),
+    BLOCKER_HEAD_SHA: (
+        "Re-fetch the live PR head with gitea_view_pr, re-pin "
+        "expected_head_sha to the current head, re-validate, then retry."
+    ),
+    BLOCKER_STALE_RUNTIME: (
+        "Restart the Gitea MCP server so it reloads master's capability "
+        "gates, re-run gitea_whoami + gitea_resolve_task_capability, then retry."
+    ),
+    BLOCKER_WORKFLOW_HASH: (
+        "Reload the canonical workflow via gitea_load_review_workflow, then "
+        "rerun the full review-merge workflow from inventory (no approve/merge replay)."
+    ),
+    BLOCKER_SOURCE_CONTAMINATION: (
+        "Stop. Session is source-contaminated. Hand off to a reconciler for "
+        "audit/clear; do not clear markers by deleting session-state files."
+    ),
+    BLOCKER_MANUAL_BYPASS: (
+        "Stop. Manual state/mtime/source-file bypass is forbidden. Use "
+        "sanctioned MCP tools only; never delete or rewrite session-state "
+        "or lock files by hand."
+    ),
+}
+
+
+def is_mutation_task(task: str | None) -> bool:
+    """True when *task* is in the #604 mutation set (normalized)."""
+    name = (task or "").strip().lower().removeprefix("gitea_")
+    if not name:
+        return False
+    if name in MUTATION_TASKS:
+        return True
+    # Accept gitea_ prefixed membership for aliases already in the set.
+    return f"gitea_{name}" in MUTATION_TASKS
+
+
+def roles_compatible(active_role: str | None, required_role: str | None) -> bool:
+    """Whether *active_role* may perform a task that maps to *required_role*.
+
+    Exact match always passes. Reconcilers may run author-class mutations
+    (comment/close/cleanup) that the capability map stamps as ``author`` —
+    matching the long-standing reconciler exemption for control-checkout
+    work. No other cross-role substitutions are allowed.
+
+    Capability-authorized multi-role tasks (e.g. reviewer holding
+    ``gitea.issue.comment`` for ``comment_issue``) are handled by
+    :func:`authorization_compatible`, not by expanding this role matrix.
+    """
+    active = (active_role or "").strip().lower()
+    required = (required_role or "").strip().lower()
+    if not active or not required:
+        return True
+    if active == required:
+        return True
+    if active == "reconciler" and required in {"author", "reconciler"}:
+        return True
+    return False
+
+
+def authorization_compatible(
+    active_role: str | None,
+    required_role: str | None,
+    *,
+    required_permission: str | None = None,
+    allowed_operations: Any = None,
+) -> bool:
+    """Whether the active session may run a task given role *and* capability.
+
+    Order:
+
+    1. :func:`roles_compatible` (exact role or narrow reconciler→author).
+    2. Capability possession: when *required_permission* is non-empty and
+       present in *allowed_operations*, allow even if the nominal task role
+       differs (e.g. reviewer/merger with ``gitea.issue.comment`` on
+       ``comment_issue`` / ``mark_issue`` / ``set_issue_labels`` /
+       ``lock_issue``).
+
+    Fail closed otherwise. This is **not** a broad reviewer→author or
+    merger→author role rewrite: without the specific permission the check
+    still denies. Missing *allowed_operations* when a permission is required
+    also fails closed (callers must supply the live profile op list).
+    """
+    if roles_compatible(active_role, required_role):
+        return True
+    perm = (required_permission or "").strip()
+    if not perm:
+        return False
+    if allowed_operations is None:
+        return False
+    try:
+        ops = {str(o).strip() for o in allowed_operations if o is not None}
+    except TypeError:
+        return False
+    return perm in ops
+
+
+def _blocker(
+    kind: str,
+    reasons: list[str],
+    *,
+    exact_next_action: str | None = None,
+    detail: dict | None = None,
+) -> dict[str, Any]:
+    if kind not in BLOCKER_KINDS:
+        raise ValueError(f"unknown anti-stomp blocker kind: {kind!r}")
+    return {
+        "kind": kind,
+        "reasons": list(reasons),
+        "exact_next_action": exact_next_action or _NEXT_ACTIONS[kind],
+        "detail": dict(detail or {}),
+    }
+
+
+def _allowed_result(checks: dict[str, Any]) -> dict[str, Any]:
+    return {
+        "allowed": True,
+        "block": False,
+        "blockers": [],
+        "reasons": [],
+        "exact_next_action": "proceed",
+        "blocker_kind": None,
+        "checks": checks,
+    }
+
+
+def _blocked_result(
+    blockers: list[dict[str, Any]],
+    checks: dict[str, Any],
+) -> dict[str, Any]:
+    primary = blockers[0]
+    all_reasons: list[str] = []
+    for b in blockers:
+        for r in b.get("reasons") or []:
+            if r not in all_reasons:
+                all_reasons.append(r)
+    return {
+        "allowed": False,
+        "block": True,
+        "blockers": blockers,
+        "reasons": all_reasons,
+        "exact_next_action": primary["exact_next_action"],
+        "blocker_kind": primary["kind"],
+        "checks": checks,
+    }
+
+
+def assess_anti_stomp_preflight(
+    *,
+    task: str | None = None,
+    # repo/org
+    remote: str | None = None,
+    resolved_org: str | None = None,
+    resolved_repo: str | None = None,
+    local_remote_url: str | None = None,
+    org_explicit: bool = False,
+    repo_explicit: bool = False,
+    check_repo: bool = True,
+    # profile/role + capability
+    profile_name: str | None = None,
+    profile_role: str | None = None,
+    required_role: str | None = None,
+    required_permission: str | None = None,
+    allowed_operations: Any = None,
+    check_role: bool = True,
+    # root checkout + worktree
+    workspace_path: str | None = None,
+    project_root: str | None = None,
+    current_branch: str | None = None,
+    root_head_sha: str | None = None,
+    root_porcelain: str | None = None,
+    remote_master_sha: str | None = None,
+    check_root_checkout: bool = True,
+    check_worktree: bool = True,
+    # stale runtime (master parity)
+    startup_head: str | None = None,
+    current_code_head: str | None = None,
+    check_stale_runtime: bool = True,
+    # lease ownership
+    lease_required: bool = False,
+    foreign_lease: bool | None = None,
+    lease_owner_session: str | None = None,
+    active_session_id: str | None = None,
+    lease_owner_identity: str | None = None,
+    active_identity: str | None = None,
+    lease_reasons: list[str] | None = None,
+    # terminal lock
+    terminal_lock_blocks: bool | None = None,
+    terminal_lock_reasons: list[str] | None = None,
+    # expected head SHA
+    require_head_sha: bool = False,
+    expected_head_sha: str | None = None,
+    live_head_sha: str | None = None,
+    # workflow hash
+    workflow_hash_valid: bool | None = None,
+    workflow_hash_reasons: list[str] | None = None,
+    # source contamination (stable-branch push / approval contamination)
+    source_contaminated: bool | None = None,
+    contamination_reasons: list[str] | None = None,
+    # manual bypass
+    manual_bypass_attempted: bool = False,
+    manual_bypass_reasons: list[str] | None = None,
+) -> dict[str, Any]:
+    """Fail-closed common anti-stomp assessment (pure).
+
+    Only checks for which facts are supplied (or which are required by the
+    mutation category) are evaluated. Unsupplied optional facts are recorded
+    as ``skipped`` so callers can see coverage without inventing defaults.
+
+    Returns a structured dict with ``allowed``/``block``, typed ``blockers``,
+    aggregated ``reasons``, ``exact_next_action``, and per-check ``checks``.
+    """
+    task_name = (task or "").strip()
+    role = (profile_role or "").strip().lower() or None
+    req_role = (required_role or "").strip().lower() or None
+    req_perm = (required_permission or "").strip() or None
+    checks: dict[str, Any] = {
+        "task": task_name or None,
+        "profile_name": profile_name,
+        "profile_role": role,
+        "required_role": req_role,
+        "required_permission": req_perm,
+    }
+    blockers: list[dict[str, Any]] = []
+
+    # ── manual bypass (always first; never skippable when attempted) ─────────
+    if manual_bypass_attempted:
+        reasons = list(manual_bypass_reasons or [
+            "manual state/mtime/source-file bypass attempt detected (fail closed)"
+        ])
+        blockers.append(_blocker(BLOCKER_MANUAL_BYPASS, reasons))
+        checks["manual_bypass"] = {"block": True, "reasons": reasons}
+    else:
+        checks["manual_bypass"] = {"block": False, "skipped": False}
+
+    # ── repo/org ─────────────────────────────────────────────────────────────
+    if check_repo and resolved_org is not None and resolved_repo is not None:
+        repo_assessment = remote_repo_guard.assess_remote_repo_match(
+            remote=remote or "",
+            resolved_org=resolved_org,
+            resolved_repo=resolved_repo,
+            local_remote_url=local_remote_url,
+            org_explicit=org_explicit,
+            repo_explicit=repo_explicit,
+        )
+        checks["repo"] = {
+            "block": bool(repo_assessment.get("block")),
+            "reasons": list(repo_assessment.get("reasons") or []),
+            "resolved_org": resolved_org,
+            "resolved_repo": resolved_repo,
+        }
+        if repo_assessment.get("block"):
+            blockers.append(
+                _blocker(
+                    BLOCKER_WRONG_REPO,
+                    list(repo_assessment.get("reasons") or ["repo/org mismatch"]),
+                    detail={
+                        "resolved_org": resolved_org,
+                        "resolved_repo": resolved_repo,
+                        "local_remote_url": local_remote_url,
+                    },
+                )
+            )
+    else:
+        checks["repo"] = {"block": False, "skipped": True}
+
+    # ── profile/role + capability ────────────────────────────────────────────
+    # Role match OR possession of the task's required permission (Blocker A).
+    # Reviewer/merger may run issue-comment-class tasks when they hold
+    # gitea.issue.comment; unauthorized escalation without the permission
+    # still fails closed.
+    if check_role and req_role and role:
+        authorized = authorization_compatible(
+            role,
+            req_role,
+            required_permission=req_perm,
+            allowed_operations=allowed_operations,
+        )
+        if not authorized:
+            perm_clause = (
+                f", required_permission='{req_perm}'" if req_perm else ""
+            )
+            reasons = [
+                f"active profile role '{role}' is not authorized for task "
+                f"'{task_name or '(unknown)'}' (required_role='{req_role}'"
+                f"{perm_clause}; profile={profile_name or '(unknown)'})"
+            ]
+            checks["role"] = {
+                "block": True,
+                "reasons": reasons,
+                "required_permission": req_perm,
+                "capability_authorized": False,
+            }
+            blockers.append(
+                _blocker(
+                    BLOCKER_WRONG_ROLE,
+                    reasons,
+                    detail={
+                        "profile_name": profile_name,
+                        "profile_role": role,
+                        "required_role": req_role,
+                        "required_permission": req_perm,
+                    },
+                )
+            )
+        else:
+            checks["role"] = {
+                "block": False,
+                "reasons": [],
+                "required_permission": req_perm,
+                "capability_authorized": bool(
+                    req_perm
+                    and allowed_operations is not None
+                    and req_perm in {
+                        str(o).strip() for o in (allowed_operations or [])
+                        if o is not None
+                    }
+                    and not roles_compatible(role, req_role)
+                ),
+            }
+    else:
+        checks["role"] = {"block": False, "skipped": True}
+
+    # ── stale runtime (master parity) ────────────────────────────────────────
+    if check_stale_runtime and (
+        startup_head is not None or current_code_head is not None
+    ):
+        parity = master_parity_gate.assess_master_parity(
+            {"startup_head": startup_head},
+            current_code_head,
+        )
+        stale_reasons = master_parity_gate.parity_block_reasons(parity)
+        checks["stale_runtime"] = {
+            "block": bool(stale_reasons),
+            "reasons": list(stale_reasons),
+            "startup_head": startup_head,
+            "current_code_head": current_code_head,
+            "stale": bool(parity.get("stale")),
+        }
+        if stale_reasons:
+            blockers.append(
+                _blocker(
+                    BLOCKER_STALE_RUNTIME,
+                    list(stale_reasons),
+                    detail={
+                        "startup_head": startup_head,
+                        "current_code_head": current_code_head,
+                    },
+                )
+            )
+    else:
+        checks["stale_runtime"] = {"block": False, "skipped": True}
+
+    # ── root checkout ────────────────────────────────────────────────────────
+    if (
+        check_root_checkout
+        and workspace_path is not None
+        and project_root is not None
+        and root_porcelain is not None
+    ):
+        root_assessment = root_checkout_guard.assess_root_checkout_guard(
+            workspace_path=workspace_path,
+            canonical_repo_root=project_root,
+            current_branch=current_branch,
+            head_sha=root_head_sha,
+            porcelain_status=root_porcelain,
+            remote_master_sha=remote_master_sha,
+            resolved_role=req_role or role,
+            actual_role=role,
+        )
+        checks["root_checkout"] = {
+            "block": bool(root_assessment.get("block")),
+            "reasons": list(root_assessment.get("reasons") or []),
+        }
+        if root_assessment.get("block"):
+            blockers.append(
+                _blocker(
+                    BLOCKER_ROOT_CHECKOUT,
+                    list(root_assessment.get("reasons") or [
+                        "control checkout is not clean master"
+                    ]),
+                    detail={
+                        "workspace_path": workspace_path,
+                        "project_root": project_root,
+                        "current_branch": current_branch,
+                    },
+                )
+            )
+    else:
+        checks["root_checkout"] = {"block": False, "skipped": True}
+
+    # ── worktree under branches (author) ─────────────────────────────────────
+    worktree_role = role or req_role
+    if (
+        check_worktree
+        and workspace_path is not None
+        and project_root is not None
+        and worktree_role in _BRANCHES_REQUIRED_ROLES
+    ):
+        wt = author_mutation_worktree.assess_author_mutation_worktree(
+            workspace_path=workspace_path,
+            project_root=project_root,
+            current_branch=current_branch,
+        )
+        checks["worktree"] = {
+            "block": bool(wt.get("block")),
+            "reasons": list(wt.get("reasons") or []),
+            "under_branches": wt.get("under_branches"),
+        }
+        if wt.get("block"):
+            blockers.append(
+                _blocker(
+                    BLOCKER_WRONG_WORKTREE,
+                    list(wt.get("reasons") or [
+                        "mutation worktree is not under branches/"
+                    ]),
+                    detail={
+                        "workspace_path": workspace_path,
+                        "project_root": project_root,
+                    },
+                )
+            )
+    else:
+        checks["worktree"] = {
+            "block": False,
+            "skipped": worktree_role not in _BRANCHES_REQUIRED_ROLES
+            or workspace_path is None,
+        }
+
+    # ── foreign lease ────────────────────────────────────────────────────────
+    lease_needed = lease_required or (
+        task_name.removeprefix("gitea_") in {
+            t.removeprefix("gitea_") for t in _LEASE_AWARE_TASKS
+        }
+        and foreign_lease is not None
+    )
+    if lease_needed or foreign_lease is True:
+        is_foreign = bool(foreign_lease)
+        if foreign_lease is None and lease_required:
+            # Ownership must be proven when lease_required; missing ownership
+            # facts fail closed.
+            owner_ok = (
+                (not lease_owner_session and not active_session_id)
+                or (
+                    lease_owner_session
+                    and active_session_id
+                    and lease_owner_session == active_session_id
+                )
+            )
+            identity_ok = (
+                (not lease_owner_identity and not active_identity)
+                or (
+                    lease_owner_identity
+                    and active_identity
+                    and lease_owner_identity == active_identity
+                )
+            )
+            if not owner_ok or not identity_ok:
+                is_foreign = True
+        reasons = list(lease_reasons or [])
+        if is_foreign and not reasons:
+            reasons = [
+                "active lease is owned by a foreign session or identity "
+                "(anti-stomp fail closed)"
+            ]
+        checks["lease"] = {
+            "block": is_foreign,
+            "reasons": reasons if is_foreign else [],
+            "lease_owner_session": lease_owner_session,
+            "active_session_id": active_session_id,
+        }
+        if is_foreign:
+            blockers.append(
+                _blocker(BLOCKER_FOREIGN_LEASE, reasons)
+            )
+    else:
+        checks["lease"] = {"block": False, "skipped": True}
+
+    # ── terminal lock ────────────────────────────────────────────────────────
+    if terminal_lock_blocks is not None:
+        reasons = list(terminal_lock_reasons or [])
+        if terminal_lock_blocks and not reasons:
+            reasons = [
+                "terminal review-decision lock is active for this head "
+                "(anti-stomp fail closed)"
+            ]
+        checks["terminal_lock"] = {
+            "block": bool(terminal_lock_blocks),
+            "reasons": reasons if terminal_lock_blocks else [],
+        }
+        if terminal_lock_blocks:
+            blockers.append(_blocker(BLOCKER_TERMINAL_LOCK, reasons))
+    else:
+        checks["terminal_lock"] = {"block": False, "skipped": True}
+
+    # ── expected head SHA ────────────────────────────────────────────────────
+    if require_head_sha or (
+        expected_head_sha is not None and live_head_sha is not None
+    ):
+        exp = (expected_head_sha or "").strip().lower()
+        live = (live_head_sha or "").strip().lower()
+        mismatch = False
+        reasons: list[str] = []
+        if require_head_sha and not exp:
+            mismatch = True
+            reasons.append(
+                "expected_head_sha is required before this mutation "
+                "(anti-stomp fail closed)"
+            )
+        elif exp and live and exp != live:
+            mismatch = True
+            reasons.append(
+                f"expected_head_sha '{exp}' does not match live PR head "
+                f"'{live}' (anti-stomp fail closed; stale prompt data)"
+            )
+        elif require_head_sha and exp and not live:
+            mismatch = True
+            reasons.append(
+                "live PR head SHA could not be determined to corroborate "
+                "expected_head_sha (anti-stomp fail closed)"
+            )
+        checks["head_sha"] = {
+            "block": mismatch,
+            "reasons": reasons,
+            "expected_head_sha": expected_head_sha,
+            "live_head_sha": live_head_sha,
+        }
+        if mismatch:
+            blockers.append(_blocker(BLOCKER_HEAD_SHA, reasons))
+    else:
+        checks["head_sha"] = {"block": False, "skipped": True}
+
+    # ── workflow hash ────────────────────────────────────────────────────────
+    if workflow_hash_valid is not None:
+        reasons = list(workflow_hash_reasons or [])
+        if not workflow_hash_valid and not reasons:
+            reasons = [
+                "workflow load proof missing or hash stale "
+                "(anti-stomp fail closed)"
+            ]
+        checks["workflow_hash"] = {
+            "block": not bool(workflow_hash_valid),
+            "reasons": reasons if not workflow_hash_valid else [],
+        }
+        if not workflow_hash_valid:
+            blockers.append(_blocker(BLOCKER_WORKFLOW_HASH, reasons))
+    else:
+        checks["workflow_hash"] = {"block": False, "skipped": True}
+
+    # ── source contamination ─────────────────────────────────────────────────
+    if source_contaminated is not None:
+        reasons = list(contamination_reasons or [])
+        if source_contaminated and not reasons:
+            reasons = [
+                "session is source-contaminated (stable-branch push or "
+                "contaminated approval path); anti-stomp fail closed"
+            ]
+        checks["source_contamination"] = {
+            "block": bool(source_contaminated),
+            "reasons": reasons if source_contaminated else [],
+        }
+        if source_contaminated:
+            blockers.append(
+                _blocker(BLOCKER_SOURCE_CONTAMINATION, reasons)
+            )
+    else:
+        checks["source_contamination"] = {"block": False, "skipped": True}
+
+    if blockers:
+        return _blocked_result(blockers, checks)
+    return _allowed_result(checks)
+
+
+def format_anti_stomp_error(assessment: dict[str, Any]) -> str:
+    """Single RuntimeError message for MCP mutation gates."""
+    kind = assessment.get("blocker_kind") or "unknown"
+    reasons = "; ".join(
+        assessment.get("reasons")
+        or ["anti-stomp preflight blocked the mutation"]
+    )
+    next_action = assessment.get("exact_next_action") or "stop and diagnose"
+    return (
+        f"Anti-stomp preflight (#604) blocked mutation "
+        f"[{kind}]: {reasons}. "
+        f"exact_next_action: {next_action}"
+    )
+
+
+def block_response(
+    assessment: dict[str, Any],
+    **extra_fields: Any,
+) -> dict[str, Any]:
+    """Structured tool-return payload for a blocked mutation."""
+    payload = {
+        "success": False,
+        "performed": False,
+        "blocked": True,
+        "anti_stomp": True,
+        "blocker_kind": assessment.get("blocker_kind"),
+        "blockers": list(assessment.get("blockers") or []),
+        "reasons": list(assessment.get("reasons") or []),
+        "exact_next_action": assessment.get("exact_next_action"),
+        "checks": assessment.get("checks") or {},
+    }
+    payload.update(extra_fields)
+    return payload
+
+
+def contamination_from_stable_marker(
+    marker: dict | None,
+    *,
+    task: str | None,
+    actual_role: str | None,
+) -> tuple[bool, list[str]]:
+    """Derive source-contamination facts from a #671 durable marker."""
+    if not marker:
+        return False, []
+    gate = stable_branch_push_guard.assess_contamination_gate(
+        marker,
+        task=task,
+        actual_role=actual_role,
+    )
+    if gate.get("block"):
+        return True, list(gate.get("reasons") or [])
+    return False, []

docs/observability/sentry-integration.md

@@ -0,0 +1,138 @@
+# Self-hosted Sentry observability for the Gitea MCP server (#606)
+
+Optional, **off-by-default** instrumentation that reports MCP runtime errors,
+fail-closed workflow blockers, lease / terminal-lock / stale-runtime
+collisions, and recurring watchdog check-ins to a **self-hosted** Sentry at
+`https://sentry.prgs.cc/`.
+
+> **Gitea remains the source of truth.** Sentry is observe-only. It never
+> approves, merges, closes, or otherwise mutates Gitea workflow state, and it
+> never bypasses leases, #332, workflow roles, or the MCP gates. Sentry alerts
+> may only feed the *sanctioned* Gitea issue/comment path via the #612 incident
+> bridge — never a direct write.
+
+Implemented by [`sentry_observability.py`](../../sentry_observability.py).
+
+---
+
+## 1. Create the Sentry project
+
+1. Sign in to the self-hosted Sentry at **`https://sentry.prgs.cc/`** (this is
+   **not** Sentry Cloud — do not use `*.ingest.sentry.io`).
+2. Create a new **Python** project named **`gitea-tools-mcp`**.
+3. Open **Settings → Projects → gitea-tools-mcp → Client Keys (DSN)** and copy
+   the DSN. It looks like `https://<publickey>@sentry.prgs.cc/<project-id>`.
+4. **Never commit the DSN.** It is a runtime secret supplied via env var only.
+
+## 2. Configure the environment
+
+All configuration is env-var driven (see [`.env.example`](../../.env.example)):
+
+| Variable | Purpose | Default |
+|----------|---------|---------|
+| `MCP_SENTRY_ENABLED` | Master gate (`1/true/yes/on`). Required. | off |
+| `SENTRY_DSN` | Self-hosted DSN. Required. | *(empty)* |
+| `SENTRY_ENVIRONMENT` | `local` / `dev` / `prod` tag. | `development` |
+| `SENTRY_RELEASE` | Release id (git SHA or version). | *(none)* |
+| `MCP_SENTRY_TRACES_SAMPLE_RATE` | Perf-trace sample rate `0.0–1.0` (clamped). | `0.0` |
+| `MCP_SENTRY_ENABLE_LOGS` | Forward Python logs as structured logs. | off |
+
+**The feature stays completely off unless `MCP_SENTRY_ENABLED` is truthy *and*
+`SENTRY_DSN` is non-empty.** With either missing, `init_sentry()` is a no-op,
+the SDK is never initialised, and no events are sent — existing tool behaviour
+and API-call patterns are unchanged.
+
+### Per-environment examples
+
+```bash
+# local (quiet: capture errors/blockers, no traces)
+export MCP_SENTRY_ENABLED=1
+export SENTRY_DSN="https://<key>@sentry.prgs.cc/<id>"
+export SENTRY_ENVIRONMENT=local
+
+# dev (light tracing + logs)
+export MCP_SENTRY_ENABLED=1
+export SENTRY_DSN="https://<key>@sentry.prgs.cc/<id>"
+export SENTRY_ENVIRONMENT=dev
+export MCP_SENTRY_TRACES_SAMPLE_RATE=0.2
+export MCP_SENTRY_ENABLE_LOGS=1
+
+# prod (errors/blockers + low-rate tracing, release-tagged)
+export MCP_SENTRY_ENABLED=1
+export SENTRY_DSN="https://<key>@sentry.prgs.cc/<id>"
+export SENTRY_ENVIRONMENT=prod
+export SENTRY_RELEASE="$(git rev-parse --short HEAD)"
+export MCP_SENTRY_TRACES_SAMPLE_RATE=0.05
+```
+
+The optional SDK is pinned in [`requirements.txt`](../../requirements.txt)
+(`sentry-sdk==2.20.0`). It is imported lazily: if the package is absent, the
+module still imports and every entry point is a safe no-op.
+
+## 3. What is instrumented
+
+| Signal | Where | Notes |
+|--------|-------|-------|
+| Startup init | `gitea_mcp_server.py` `__main__`, before `mcp.run` | Prints a redaction-safe status line to stderr. |
+| Failing mutations (exceptions) | `_audited(...)` context manager | `capture_exception` with scrubbed tags. |
+| Fail-closed blockers / failed mutations | `_audit_pr_result(...)` (BLOCKED/FAILED) | Structured `capture_workflow_blocker` event incl. the canonical next action when available (criterion 7). |
+| Allocator watchdog check-ins | `gitea_allocate_next_work` tool | `allocator_health`, `stale_lease_scan`, `terminal_lock_scan`. |
+| Namespace-health check-in | `gitea_assess_mcp_namespace_health` tool | `namespace_health`. |
+
+All capture paths are **best-effort / fail open**: a Sentry outage or capture
+error never breaks an MCP tool success path.
+
+## 4. Cron / watchdog monitors
+
+`sentry_observability.MONITOR_SLUGS` defines stable check-in slugs:
+
+| Registry key | Sentry monitor slug | Wired at |
+|--------------|--------------------|----------|
+| `stale_lease_scan` | `gitea-mcp-stale-lease-scan` | allocator run (global lease expiry) |
+| `terminal_lock_scan` | `gitea-mcp-terminal-lock-scan` | allocator run (terminal-lock lookup) |
+| `allocator_health` | `gitea-mcp-allocator-health` | allocator run |
+| `namespace_health` | `gitea-mcp-namespace-health` | namespace-health probe |
+| `dashboard_freshness` | `gitea-mcp-dashboard-freshness` | call `monitor_checkin("dashboard_freshness", ...)` from the dashboard refresh job (#605) |
+| `reconciler_cleanup` | `gitea-mcp-reconciler-cleanup` | call `monitor_checkin("reconciler_cleanup", ...)` from the reconciler cleanup entrypoint |
+
+Create matching Cron monitors in Sentry with those slugs. Emit an
+`in_progress` check-in at job start and `ok`/`error` at completion via
+`sentry_observability.monitor_checkin(slug_key, status)`.
+
+## 5. Redaction guarantees (fail closed)
+
+Redaction fails *closed*: if a field cannot be proven safe it is dropped rather
+than sent. The `before_send` (and `before_send_log`) hook `scrub_event`
+recursively redacts every outgoing event; on any error it drops the event
+entirely. Guarantees, proven by `tests/test_sentry_observability.py`:
+
+- **No** tokens, passwords, keychain IDs, DSNs, cookies, or `user:pass@host`.
+- **No** raw session-state or full prompt/comment bodies — `session_id` is only
+  ever surfaced as a 12-char `session_id_hash`.
+- **No** private config contents or raw credential headers.
+- **No** full local filesystem paths — a worktree path collapses to a coarse
+  `worktree_category` (`author` / `reviewer` / `merger` / `reconciler` /
+  `branches` / `root` / `other`).
+- Only the allowlisted tag keys in `ALLOWED_TAG_KEYS` are ever attached.
+
+## 6. Coexistence with GlitchTip / the #612 incident bridge
+
+This is the **outbound** path (MCP → Sentry SDK). It complements — it does not
+replace — the **inbound** [`incident_bridge.py`](../../incident_bridge.py)
+(#612), which turns Sentry/GlitchTip *observations* into durable Gitea issues
+and `incident_links` rows.
+
+- Prefer **one** observability path per environment. Point the MCP server's
+  `SENTRY_DSN` at the same self-hosted `gitea-tools-mcp` project that the #612
+  bridge reconciles from, so an MCP-reported error and its Gitea issue line up.
+- GlitchTip is Sentry-protocol compatible; if an existing GlitchTip DSN is in
+  use, either migrate it to `https://sentry.prgs.cc/` or document the split
+  (MCP → Sentry, legacy → GlitchTip) explicitly for operators.
+- The bridge remains the **only** sanctioned route from an alert back into
+  Gitea workflow state.
+
+## 7. Non-goals
+
+- Sentry must **not** become the workflow source of truth.
+- Sentry must **not** approve, merge, close, or mutate Gitea workflow state.
+- Sentry must **not** bypass leases, #332, workflow roles, or the MCP gates.

mcp_session_state.py

@@ -36,6 +36,15 @@ KIND_REVIEW_DRAFT = "review_draft"
 # other session proofs; a contaminated session fails closed on gated mutations
 # until a reconciler audits and clears it.
 KIND_STABLE_BRANCH_CONTAMINATION = "stable_branch_contamination"
+# Durable shadow of the in-memory reviewer session lease (#702). Written on
+# every sanctioned record/heartbeat and removed on sanctioned clear, so a
+# daemon that dies without teardown leaves provable orphan evidence (owner
+# pid + session id) for the guarded lease-cleanup path. Never a substitute
+# for the in-session lease: mutation gates ignore it.
+KIND_REVIEWER_SESSION_LEASE = "reviewer_session_lease"
+# Durable audit record written whenever the daemon's sanctioned stale-binding
+# recovery clears an inherited GITEA_ACTIVE_WORKTREE binding (#702).
+KIND_STALE_BINDING_RECOVERY = "stale_binding_recovery"
 # #709: archive prior terminal decision ledgers instead of silent overwrite.
 KIND_DECISION_LOCK_ARCHIVE = "review_decision_lock_archive"
 # #709: post-merge cleanup/audit reconciliation-required durable record.
@@ -59,6 +68,9 @@ RECOVERY_CRITICAL_KINDS = frozenset(
         KIND_POST_MERGE_DECISION_RECOVERY,
         KIND_IRRECOVERABLE_DECISION_PROVENANCE,
         KIND_IRRECOVERABLE_PROVENANCE_AUTH,
+        # #702 crash-orphan evidence (must outlive TTL; F4)
+        KIND_REVIEWER_SESSION_LEASE,
+        KIND_STALE_BINDING_RECOVERY,
     }
 )
 
@@ -138,6 +150,7 @@ def state_key(
     org: str | None = None,
     repo: str | None = None,
     profile_identity: str | None = None,
+    instance_id: str | None = None,
 ) -> str:
     """Build durable filename key.
 
@@ -145,17 +158,20 @@ def state_key(
     so the primary key is kind + profile identity. Remote/org/repo are stored
     inside the payload and validated on load (#559), which lets a later daemon
     process recover state without already knowing the remote argument.
+
+    *instance_id* extends the key for kinds where one-per-profile collides —
+    concurrent same-profile sessions each own their reviewer-lease shadow
+    (#702 F5), keyed by their lease session id so a later heartbeat can never
+    overwrite or misattribute another session's crash evidence.
     """
     # Keep remote/org/repo parameters for API stability / future kinds; they are
     # intentionally not part of the filename for session-scoped proofs.
     _ = (remote, org, repo)
-    return "-".join(
-        _sanitize_segment(part)
-        for part in (
-            kind,
-            profile_identity or "unknown-profile",
-        )
-    )
+    parts = [kind, profile_identity or "unknown-profile"]
+    instance = (instance_id or "").strip()
+    if instance:
+        parts.append(instance)
+    return "-".join(_sanitize_segment(part) for part in parts)
 
 
 def state_file_path(
@@ -166,6 +182,7 @@ def state_file_path(
     repo: str | None = None,
     profile_identity: str | None = None,
     state_dir: str | None = None,
+    instance_id: str | None = None,
 ) -> str:
     root = (state_dir or default_state_dir()).strip()
     name = state_key(
@@ -174,6 +191,7 @@ def state_file_path(
         org=org,
         repo=repo,
         profile_identity=profile_identity,
+        instance_id=instance_id,
     )
     return os.path.join(root, f"{name}.json")
 
@@ -430,6 +448,7 @@ def load_state(
     repo: str | None = None,
     profile_identity: str | None = None,
     state_dir: str | None = None,
+    instance_id: str | None = None,
 ) -> dict[str, Any] | None:
     """Load durable state payload when identity checks pass."""
     profile = current_profile_identity(profile_identity=profile_identity)
@@ -440,6 +459,7 @@ def load_state(
         repo=repo,
         profile_identity=profile,
         state_dir=state_dir,
+        instance_id=instance_id,
     )
     lock_path = f"{path}.lock"
     with _exclusive_file_lock(lock_path):
@@ -485,6 +505,7 @@ def save_state(
     repo: str | None = None,
     profile_identity: str | None = None,
     state_dir: str | None = None,
+    instance_id: str | None = None,
 ) -> dict[str, Any] | None:
     """Persist or clear durable state for the given session identity key."""
     profile = current_profile_identity(
@@ -497,6 +518,11 @@ def save_state(
     key_remote = remote if remote is not None else (payload or {}).get("remote")
     key_org = org if org is not None else (payload or {}).get("org")
     key_repo = repo if repo is not None else (payload or {}).get("repo")
+    key_instance = (
+        (instance_id or "").strip()
+        or str((payload or {}).get("instance_id") or "").strip()
+        or None
+    )
 
     root = _ensure_state_dir(state_dir)
     path = state_file_path(
@@ -506,6 +532,7 @@ def save_state(
         repo=key_repo,
         profile_identity=profile,
         state_dir=root,
+        instance_id=key_instance,
     )
     lock_path = f"{path}.lock"
     with _exclusive_file_lock(lock_path):
@@ -535,6 +562,8 @@ def save_state(
             body["repo"] = key_repo
         # Stamp session-state authority used for this write (#695 AC2 / AC6).
         body.setdefault("session_state_dir", root)
+        if key_instance:
+            body["instance_id"] = key_instance
         try:
             import mcp_daemon_guard
 
@@ -568,6 +597,8 @@ def save_state(
             "transport": body.get("transport"),
             "payload": body,
         }
+        if key_instance:
+            envelope["instance_id"] = key_instance
         _write_json(path, envelope)
         return dict(body)
 
@@ -580,6 +611,7 @@ def clear_state(
     repo: str | None = None,
     profile_identity: str | None = None,
     state_dir: str | None = None,
+    instance_id: str | None = None,
 ) -> None:
     save_state(
         kind=kind,
@@ -589,8 +621,59 @@ def clear_state(
         repo=repo,
         profile_identity=profile_identity,
         state_dir=state_dir,
+        instance_id=instance_id,
     )
 
+def list_states(
+    *,
+    kind: str,
+    profile_identity: str | None = None,
+    state_dir: str | None = None,
+) -> list[dict[str, Any]]:
+    """Enumerate durable records of *kind* across all instance keys.
+
+    Crash recovery cannot know which session id left a shadow behind, so it
+    must be able to enumerate every record of a kind (#702 F5). Identity
+    checks (profile / TTL policy) apply per record exactly as in
+    :func:`load_state`; records that fail closed are omitted.
+    """
+    root = (state_dir or default_state_dir()).strip()
+    if not root or not os.path.isdir(root):
+        return []
+    profile = current_profile_identity(profile_identity=profile_identity)
+    results: list[dict[str, Any]] = []
+    try:
+        names = sorted(os.listdir(root))
+    except OSError:
+        return []
+    for name in names:
+        if not name.endswith(".json") or name.startswith("."):
+            continue
+        envelope = _read_json(os.path.join(root, name))
+        if not envelope or envelope.get("kind") != kind:
+            continue
+        payload = envelope.get("payload")
+        if not isinstance(payload, dict):
+            continue
+        merged = dict(payload)
+        for key in (
+            "kind",
+            "remote",
+            "org",
+            "repo",
+            "profile_identity",
+            "session_profile_lock",
+            "recorded_at",
+            "updated_at",
+            "writer_pid",
+        ):
+            if key in envelope and key not in merged:
+                merged[key] = envelope[key]
+        if identity_match_reasons(merged, profile_identity=profile):
+            continue
+        results.append(merged)
+    return results
+
 def list_decision_lock_profile_identities(
     state_dir: str | None = None,
 ) -> list[str]:

merger_lease_adoption.py

@@ -18,11 +18,13 @@ DEFAULT_ADOPTION_REASON = "merger-handoff-approved-head"
 
 SOURCE_ADOPT = "gitea_adopt_merger_pr_lease"
 SOURCE_ACQUIRE = "gitea_acquire_reviewer_pr_lease"
+SOURCE_ACQUIRE_MERGER = "gitea_acquire_merger_pr_lease"
 SOURCE_HEARTBEAT = "gitea_heartbeat_reviewer_pr_lease"
 
 SANCTIONED_PROVENANCE_SOURCES = frozenset({
     SOURCE_ADOPT,
     SOURCE_ACQUIRE,
+    SOURCE_ACQUIRE_MERGER,
     SOURCE_HEARTBEAT,
 })
 
@@ -209,8 +211,10 @@ _SANCTIONED_LEASE_EVIDENCE_RE = re.compile(
     r"(?is)\b("
     r"gitea_adopt_merger_pr_lease|"
     r"gitea_acquire_reviewer_pr_lease|"
+    r"gitea_acquire_merger_pr_lease|"
     r"lease_proof_source\s*[:=]\s*gitea_adopt_merger_pr_lease|"
     r"lease_proof_source\s*[:=]\s*gitea_acquire_reviewer_pr_lease|"
+    r"lease_proof_source\s*[:=]\s*gitea_acquire_merger_pr_lease|"
     r"lease_proof_kind\s*[:=]\s*sanctioned_adoption|"
     r"lease_proof_kind\s*[:=]\s*sanctioned_acquire|"
     r"adoption_comment_id\s*[:=]|"

namespace_workspace_binding.py

@@ -56,23 +56,46 @@ def resolve_namespace_workspace(
     env: dict[str, str] | os._Environ | None = None,
     session_lease_worktree: str | None = None,
     profile_name: str | None = None,
+    demotions: list[str] | None = None,
+    verify_paths: bool = False,
 ) -> tuple[str, str]:
-    """Return ``(resolved_path, binding_source)`` for *role_kind*."""
+    """Return ``(resolved_path, binding_source)`` for *role_kind*.
+
+    With *verify_paths*, env-sourced candidates whose path no longer exists
+    are demoted (#702): a binding to a deleted worktree can never name a
+    valid task workspace, so resolution falls through to the next candidate.
+    Explicit arguments are never demoted — a caller-declared path must fail
+    loudly downstream rather than silently rebind. Demotion notes are
+    appended to *demotions* when provided. Runtime-context and mutation
+    guards resolve through :func:`resolve_namespace_mutation_context`, which
+    always verifies.
+    """
     env_map = env if env is not None else os.environ
     role = normalize_role_kind(role_kind, profile_name=profile_name)
     role_env_key = ROLE_WORKTREE_ENVS[role]
 
-    for candidate, source in (
-        (worktree_path, "worktree_path argument"),
-        (worktree, "worktree argument"),
-        (_env_value(env_map, ACTIVE_WORKTREE_ENV), f"{ACTIVE_WORKTREE_ENV} environment variable"),
-        (_env_value(env_map, role_env_key), f"{role_env_key} environment variable"),
+    for candidate, source, env_sourced in (
+        (worktree_path, "worktree_path argument", False),
+        (worktree, "worktree argument", False),
+        (_env_value(env_map, ACTIVE_WORKTREE_ENV),
+         f"{ACTIVE_WORKTREE_ENV} environment variable", True),
+        (_env_value(env_map, role_env_key),
+         f"{role_env_key} environment variable", True),
         (session_lease_worktree if role in {"reviewer", "merger"} else None,
-         "reviewer PR lease worktree"),
+         "reviewer PR lease worktree", False),
     ):
         text = (candidate or "").strip()
-        if text:
-            return os.path.realpath(os.path.abspath(text)), source
+        if not text:
+            continue
+        real = os.path.realpath(os.path.abspath(text))
+        if verify_paths and env_sourced and not os.path.isdir(real):
+            if demotions is not None:
+                demotions.append(
+                    f"{source} '{real}' demoted: path no longer exists "
+                    "(stale binding, #702)"
+                )
+            continue
+        return real, source
 
     return os.path.realpath(process_project_root), "MCP server process root (default)"
 
@@ -88,6 +111,7 @@ def resolve_namespace_mutation_context(
     profile_name: str | None = None,
 ) -> dict:
     """Shared workspace resolution for runtime_context and mutation guards."""
+    demotions: list[str] = []
     workspace, binding_source = resolve_namespace_workspace(
         role_kind=role_kind,
         worktree_path=worktree_path,
@@ -96,6 +120,8 @@ def resolve_namespace_mutation_context(
         env=env,
         session_lease_worktree=session_lease_worktree,
         profile_name=profile_name,
+        demotions=demotions,
+        verify_paths=True,
     )
     process_root = os.path.realpath(process_project_root)
     role = normalize_role_kind(role_kind, profile_name=profile_name)
@@ -111,7 +137,7 @@ def resolve_namespace_mutation_context(
         "workspace_path": workspace,
         "workspace_binding_source": binding_source,
         "workspace_role_kind": role,
-        "ignored_bindings": pollution.get("ignored_bindings") or [],
+        "ignored_bindings": demotions + (pollution.get("ignored_bindings") or []),
         "process_project_root": process_root,
         "canonical_repo_root": canonical_root,
         "roots_aligned": canonical_root == process_root,

pr_sync_status.py

@@ -0,0 +1,648 @@
+"""PR synchronization and conflict-remediation lifecycle (#PR-SYNC).
+
+Pure assessment helpers for:
+
+* ``gitea_assess_pr_sync_status`` — recommend the next sanctioned action for an
+  open PR when the base branch advances, approvals go stale, or conflicts appear.
+* ``gitea_update_pr_branch_by_merge`` preflight — author-only, fail-closed pin
+  of both PR head and live base head; never rebase/force-push.
+
+All recommendation logic is hermetic (no network) so unit tests cover every
+acceptance path without Gitea credentials.
+"""
+
+from __future__ import annotations
+
+import re
+from typing import Any
+
+_FULL_SHA = re.compile(r"^[0-9a-f]{40}$", re.IGNORECASE)
+
+# Recommended next actions (controller / skill vocabulary).
+ACTION_MERGE_NOW = "merge_now"
+ACTION_UPDATE_BRANCH_BY_MERGE = "update_branch_by_merge"
+ACTION_AUTHOR_CONFLICT_REMEDIATION = "author_conflict_remediation"
+ACTION_FRESH_REVIEW_REQUIRED = "fresh_review_required"
+ACTION_BLOCKED = "blocked"
+
+_VALID_ACTIONS = frozenset({
+    ACTION_MERGE_NOW,
+    ACTION_UPDATE_BRANCH_BY_MERGE,
+    ACTION_AUTHOR_CONFLICT_REMEDIATION,
+    ACTION_FRESH_REVIEW_REQUIRED,
+    ACTION_BLOCKED,
+})
+
+# Author-only mutation; reviewer/merger must never update author branches.
+_AUTHOR_UPDATE_ROLES = frozenset({"author"})
+_DENIED_UPDATE_ROLES = frozenset({"reviewer", "merger", "reconciler", "mixed", "limited"})
+
+# Allowed Gitea update style — merge only (never rebase).
+UPDATE_STYLE_MERGE = "merge"
+_FORBIDDEN_UPDATE_STYLES = frozenset({"rebase", "rebase-merge", "squash", "force"})
+
+
+def _normalize_sha(value: str | None) -> str | None:
+    text = (value or "").strip().lower()
+    if not text:
+        return None
+    return text if _FULL_SHA.match(text) else None
+
+
+def _normalize_role(role: str | None) -> str:
+    return (role or "").strip().lower() or "unknown"
+
+
+def assess_pr_sync_status(
+    *,
+    host: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+    pr_number: int,
+    pr_state: str | None = None,
+    source_branch: str | None = None,
+    pr_head_sha: str | None = None,
+    base_head_sha: str | None = None,
+    commits_behind: int | None = None,
+    mergeable: bool | None = None,
+    has_conflicts: bool | None = None,
+    branch_protection_requires_current_base: bool | None = None,
+    approval_at_current_head: bool | None = None,
+    checks_status: str | None = None,
+    active_author_lock: bool | None = None,
+    active_reviewer_lease: bool | None = None,
+    active_merger_lease: bool | None = None,
+    prepared_verdict_head_sha: str | None = None,
+    checks_required: bool = True,
+) -> dict[str, Any]:
+    """Recommend the next sanctioned PR lifecycle action.
+
+    Decision order (fail closed):
+
+    1. Incomplete identity / open state / head SHAs → ``blocked``
+    2. Conflicts (or mergeable false with conflict signal) →
+       ``author_conflict_remediation``
+    3. Head changed after approval / prepared verdict for former head →
+       ``fresh_review_required`` (unless conflicts already routed author work)
+    4. Behind live base + branch protection requires current base +
+       auto-mergeable → ``update_branch_by_merge``
+    5. Approval at current head + mergeable + checks ok (+ update not
+       required) → ``merge_now``
+    6. Otherwise ``blocked`` with structured reasons
+    """
+    reasons: list[str] = []
+    pr_head = _normalize_sha(pr_head_sha)
+    base_head = _normalize_sha(base_head_sha)
+    prepared_head = _normalize_sha(prepared_verdict_head_sha)
+    state = (pr_state or "").strip().lower()
+    checks = (checks_status or "unknown").strip().lower()
+    behind = commits_behind if isinstance(commits_behind, int) and commits_behind >= 0 else None
+    requires_current = bool(branch_protection_requires_current_base)
+    approval_ok = bool(approval_at_current_head)
+
+    # Derive conflict when caller only supplies mergeable=False.
+    if has_conflicts is None and mergeable is False:
+        has_conflicts = True
+    conflicts = bool(has_conflicts) if has_conflicts is not None else None
+
+    result: dict[str, Any] = {
+        "host": (host or "").strip() or None,
+        "org": (org or "").strip() or None,
+        "repo": (repo or "").strip() or None,
+        "pr_number": pr_number,
+        "pr_state": state or None,
+        "source_branch": (source_branch or "").strip() or None,
+        "pr_head_sha": pr_head,
+        "base_head_sha": base_head,
+        "commits_behind": behind,
+        "mergeable": mergeable,
+        "has_conflicts": conflicts,
+        "branch_protection_requires_current_base": requires_current,
+        "approval_at_current_head": approval_ok if approval_at_current_head is not None else None,
+        "checks_status": checks,
+        "active_locks_and_leases": {
+            "author_lock": bool(active_author_lock) if active_author_lock is not None else None,
+            "reviewer_lease": bool(active_reviewer_lease) if active_reviewer_lease is not None else None,
+            "merger_lease": bool(active_merger_lease) if active_merger_lease is not None else None,
+        },
+        "prepared_verdict_head_sha": prepared_head,
+        "recommended_next_action": ACTION_BLOCKED,
+        "approval_valid_for_merge": False,
+        "stale_approval": False,
+        "stale_prepared_verdict": False,
+        "reasons": reasons,
+        "success": True,
+        "performed": False,
+    }
+
+    if not isinstance(pr_number, int) or pr_number <= 0:
+        reasons.append("pr_number must be a positive integer (fail closed)")
+        return result
+
+    if state and state not in ("open",):
+        reasons.append(f"PR is not open (state={state}); cannot synchronize or merge")
+        result["recommended_next_action"] = ACTION_BLOCKED
+        return result
+    if not state:
+        reasons.append("PR state missing (fail closed)")
+        return result
+
+    if pr_head is None:
+        reasons.append(
+            "exact PR head SHA missing or not a full 40-char hex SHA (fail closed)"
+        )
+    if base_head is None:
+        reasons.append(
+            "exact live base head SHA missing or not a full 40-char hex SHA (fail closed)"
+        )
+    if behind is None:
+        reasons.append("commits_behind missing or invalid (fail closed)")
+    if mergeable is None:
+        reasons.append("mergeable signal missing (fail closed)")
+    if approval_at_current_head is None:
+        reasons.append("approval_at_current_head missing (fail closed)")
+    if branch_protection_requires_current_base is None:
+        reasons.append(
+            "branch_protection_requires_current_base missing (fail closed)"
+        )
+
+    if reasons:
+        result["recommended_next_action"] = ACTION_BLOCKED
+        return result
+
+    # Stale prepared verdict / approval across heads.
+    stale_prepared = bool(prepared_head and prepared_head != pr_head)
+    result["stale_prepared_verdict"] = stale_prepared
+    stale_approval = not approval_ok
+    result["stale_approval"] = stale_approval
+    result["approval_valid_for_merge"] = bool(approval_ok and not stale_prepared)
+
+    # ── Conflicts: author remediation in dedicated worktree ──────────────
+    if conflicts is True or mergeable is False:
+        if conflicts is True or mergeable is False:
+            # Distinguish pure non-mergeable without explicit conflict flag
+            # still routes author remediation (Gitea cannot auto-update).
+            reasons.append(
+                f"PR #{pr_number} has conflicts or is not mergeable at head "
+                f"{pr_head}; route author conflict remediation in the existing "
+                "issue/PR worktree under branches/ (never force-push or rebase)"
+            )
+            if stale_approval:
+                reasons.append(
+                    "any approval at a former head is invalid; after remediation "
+                    "require a fresh independent review at the new exact head"
+                )
+            result["recommended_next_action"] = ACTION_AUTHOR_CONFLICT_REMEDIATION
+            result["approval_valid_for_merge"] = False
+            return result
+
+    # ── Stale approval / prepared verdict at former head ─────────────────
+    if not approval_ok or stale_prepared:
+        if behind and behind > 0 and requires_current and mergeable is True:
+            # Must update first; update will also invalidate approval.
+            reasons.append(
+                f"PR is {behind} commit(s) behind live base and branch protection "
+                "requires current base; automatic merge-from-base is available"
+            )
+            reasons.append(
+                "approval is not valid at the current head (or prepared verdict "
+                "is head-scoped to a former SHA); after update require fresh review"
+            )
+            result["recommended_next_action"] = ACTION_UPDATE_BRANCH_BY_MERGE
+            result["approval_valid_for_merge"] = False
+            return result
+        reasons.append(
+            "approval is not valid at the exact current PR head "
+            f"({pr_head}); never reuse a former-head approval for merge"
+        )
+        if stale_prepared:
+            reasons.append(
+                f"prepared verdict pinned to former head {prepared_head} cannot "
+                f"authorize review/merge at current head {pr_head}"
+            )
+        if active_reviewer_lease:
+            reasons.append(
+                "active reviewer lease must be released or superseded for the "
+                "new head before a fresh independent review"
+            )
+        if active_merger_lease:
+            reasons.append(
+                "active merger lease cannot cross heads; release or re-acquire "
+                "at the new exact head"
+            )
+        result["recommended_next_action"] = ACTION_FRESH_REVIEW_REQUIRED
+        result["approval_valid_for_merge"] = False
+        return result
+
+    # ── Outdated + protection requires current base, auto-updatable ──────
+    if behind and behind > 0 and requires_current:
+        if mergeable is True and conflicts is not True:
+            reasons.append(
+                f"PR is {behind} commit(s) behind live base {base_head}; "
+                "branch protection requires the PR branch to contain current base; "
+                "Gitea can merge base into the PR branch without conflicts"
+            )
+            reasons.append(
+                "route author-only update_branch_by_merge; never rebase or force-push; "
+                "new head invalidates prior approval and requires fresh independent review"
+            )
+            result["recommended_next_action"] = ACTION_UPDATE_BRANCH_BY_MERGE
+            # Approval today is at old head that will change — not mergeable yet
+            # for final merge until re-reviewed, but assess marks update path.
+            result["approval_valid_for_merge"] = False
+            result["stale_approval"] = True  # will become stale after update
+            return result
+        reasons.append(
+            "update required by branch protection but automatic merge is not available"
+        )
+        result["recommended_next_action"] = ACTION_BLOCKED
+        return result
+
+    # ── Checks gate for merge_now ────────────────────────────────────────
+    if checks_required and checks not in ("success", "passed", "ok", "none", "skipped", "not_required"):
+        if checks in ("pending", "running", "queued"):
+            reasons.append(f"required checks are not finished (status={checks})")
+            result["recommended_next_action"] = ACTION_BLOCKED
+            return result
+        if checks in ("failure", "failed", "error", "cancelled"):
+            reasons.append(f"required checks failed (status={checks})")
+            result["recommended_next_action"] = ACTION_BLOCKED
+            return result
+        # unknown — fail closed when checks_required
+        if checks == "unknown":
+            reasons.append("checks status unknown (fail closed)")
+            result["recommended_next_action"] = ACTION_BLOCKED
+            return result
+
+    # ── Ready to merge without update ────────────────────────────────────
+    # Includes: current with approval; outdated when update is NOT required.
+    if approval_ok and mergeable is True and conflicts is not True:
+        if behind and behind > 0 and not requires_current:
+            reasons.append(
+                f"PR is {behind} commit(s) behind live base but branch protection "
+                "does not require the latest base; preserve the approved head"
+            )
+        else:
+            reasons.append(
+                "PR has a valid approval at the exact current head, is conflict-free "
+                "and mergeable; do not update the branch unnecessarily"
+            )
+        reasons.append("route directly to the sanctioned merger workflow (merge_now)")
+        result["recommended_next_action"] = ACTION_MERGE_NOW
+        result["approval_valid_for_merge"] = True
+        return result
+
+    reasons.append("no sanctioned next action matched the observed PR state (fail closed)")
+    result["recommended_next_action"] = ACTION_BLOCKED
+    return result
+
+
+def assess_update_pr_branch_preflight(
+    *,
+    role_kind: str | None,
+    expected_pr_head_sha: str | None,
+    live_pr_head_sha: str | None,
+    expected_base_head_sha: str | None,
+    live_base_head_sha: str | None,
+    has_conflicts: bool | None = None,
+    mergeable: bool | None = None,
+    style: str | None = UPDATE_STYLE_MERGE,
+    has_author_lock: bool | None = None,
+    worktree_path: str | None = None,
+    worktree_owns_source_branch: bool | None = None,
+    control_checkout_clean: bool | None = None,
+    master_parity_ok: bool | None = None,
+    force_push: bool = False,
+    rebase: bool = False,
+) -> dict[str, Any]:
+    """Author-only preflight for update-by-merge; fail closed on any race.
+
+    When conflicts exist, returns a structured author-remediation handoff and
+    sets ``mutation_allowed=False`` so no partial remote update is performed.
+    """
+    reasons: list[str] = []
+    role = _normalize_role(role_kind)
+    exp_pr = _normalize_sha(expected_pr_head_sha)
+    live_pr = _normalize_sha(live_pr_head_sha)
+    exp_base = _normalize_sha(expected_base_head_sha)
+    live_base = _normalize_sha(live_base_head_sha)
+    style_norm = (style or "").strip().lower() or UPDATE_STYLE_MERGE
+
+    conflicts = has_conflicts
+    if conflicts is None and mergeable is False:
+        conflicts = True
+
+    result: dict[str, Any] = {
+        "mutation_allowed": False,
+        "performed": False,
+        "style": style_norm,
+        "role_kind": role,
+        "expected_pr_head_sha": exp_pr,
+        "live_pr_head_sha": live_pr,
+        "expected_base_head_sha": exp_base,
+        "live_base_head_sha": live_base,
+        "head_race": False,
+        "base_race": False,
+        "has_conflicts": conflicts,
+        "author_remediation_handoff": False,
+        "approval_invalidated_for_former_head": False,
+        "force_push": bool(force_push),
+        "rebase": bool(rebase),
+        "reasons": reasons,
+        "success": True,
+    }
+
+    # Role gate — author only.
+    if role not in _AUTHOR_UPDATE_ROLES:
+        reasons.append(
+            f"role '{role}' cannot update an author PR branch "
+            "(author-only; reviewer/merger denial)"
+        )
+        return result
+
+    if force_push:
+        reasons.append("force-push is forbidden for PR branch update (fail closed)")
+        return result
+    if rebase or style_norm in _FORBIDDEN_UPDATE_STYLES:
+        reasons.append(
+            f"update style '{style_norm}' forbidden; only style=merge is allowed "
+            "(never rebase)"
+        )
+        return result
+    if style_norm != UPDATE_STYLE_MERGE:
+        reasons.append(
+            f"unknown update style '{style_norm}'; expected '{UPDATE_STYLE_MERGE}'"
+        )
+        return result
+
+    if not exp_pr or not live_pr:
+        reasons.append(
+            "expected and live PR head SHAs must be full 40-char hex (fail closed)"
+        )
+    if not exp_base or not live_base:
+        reasons.append(
+            "expected and live base head SHAs must be full 40-char hex (fail closed)"
+        )
+    if reasons:
+        return result
+
+    if exp_pr != live_pr:
+        result["head_race"] = True
+        reasons.append(
+            f"PR head race: expected {exp_pr} but live is {live_pr} "
+            "(fail closed; no partial mutation)"
+        )
+        return result
+    if exp_base != live_base:
+        result["base_race"] = True
+        reasons.append(
+            f"base head race: expected {exp_base} but live is {live_base} "
+            "(fail closed; no partial mutation)"
+        )
+        return result
+
+    if has_author_lock is not True:
+        reasons.append(
+            "existing issue/PR lock required before update_branch_by_merge (fail closed)"
+        )
+    wt = (worktree_path or "").strip()
+    if not wt:
+        reasons.append("existing PR worktree path required under branches/ (fail closed)")
+    else:
+        norm = wt.replace("\\", "/")
+        if "/branches/" not in norm and not norm.startswith("branches/"):
+            reasons.append(
+                f"worktree_path '{wt}' must be under branches/ (fail closed)"
+            )
+    if worktree_owns_source_branch is False:
+        reasons.append(
+            "worktree does not own the PR source branch (fail closed)"
+        )
+    if control_checkout_clean is False:
+        reasons.append("control checkout is not clean (fail closed)")
+    if master_parity_ok is False:
+        reasons.append("runtime/master parity failed (fail closed)")
+
+    if conflicts is True:
+        result["author_remediation_handoff"] = True
+        reasons.append(
+            "conflicts present: return structured author-remediation handoff "
+            "without creating a partial remote update"
+        )
+        reasons.append(
+            "resolve conflicts explicitly in the existing dedicated worktree, "
+            "run focused and regression tests, commit/push via sanctioned author "
+            "workflow, then route the new exact head to independent review"
+        )
+        return result
+
+    if mergeable is False and conflicts is not False:
+        result["author_remediation_handoff"] = True
+        reasons.append(
+            "PR is not mergeable; refuse automatic update and hand off to "
+            "author conflict remediation (fail closed)"
+        )
+        return result
+
+    if reasons:
+        return result
+
+    result["mutation_allowed"] = True
+    reasons.append(
+        "preflight passed: author may merge live base into the PR branch via "
+        "native MCP update (style=merge only)"
+    )
+    return result
+
+
+def assess_post_update_head_transition(
+    *,
+    former_pr_head_sha: str | None,
+    new_pr_head_sha: str | None,
+    former_approval_head_sha: str | None = None,
+    former_reviewer_lease_head_sha: str | None = None,
+    former_merger_lease_head_sha: str | None = None,
+    prepared_verdict_head_sha: str | None = None,
+) -> dict[str, Any]:
+    """After a successful update-by-merge, invalidate cross-head artifacts.
+
+    The new head never inherits approval, reviewer/merger leases, or prepared
+    verdicts from the former head.
+    """
+    former = _normalize_sha(former_pr_head_sha)
+    new = _normalize_sha(new_pr_head_sha)
+    reasons: list[str] = []
+    result: dict[str, Any] = {
+        "former_pr_head_sha": former,
+        "new_pr_head_sha": new,
+        "head_changed": bool(former and new and former != new),
+        "approval_invalidated": False,
+        "reviewer_lease_superseded": False,
+        "merger_lease_superseded": False,
+        "prepared_verdict_invalidated": False,
+        "recommended_next_action": ACTION_BLOCKED,
+        "reasons": reasons,
+        "success": True,
+    }
+
+    if not former or not new:
+        reasons.append("former and new PR head SHAs required (fail closed)")
+        return result
+    if former == new:
+        reasons.append(
+            "PR head unchanged after update; unexpected for merge-from-base"
+        )
+        result["recommended_next_action"] = ACTION_BLOCKED
+        return result
+
+    result["head_changed"] = True
+    # Approval at former head is always void at new head.
+    former_approval = _normalize_sha(former_approval_head_sha) or former
+    if former_approval != new:
+        result["approval_invalidated"] = True
+        reasons.append(
+            f"approval for former head {former_approval} is invalid at new head "
+            f"{new}; never preserve approval across a head change"
+        )
+
+    rev_lease = _normalize_sha(former_reviewer_lease_head_sha)
+    if rev_lease and rev_lease != new:
+        result["reviewer_lease_superseded"] = True
+        reasons.append(
+            f"reviewer lease for head {rev_lease} cannot cross to {new}; "
+            "release or supersede before fresh review"
+        )
+    elif rev_lease is None:
+        # Unknown lease head still must not authorize at new head.
+        result["reviewer_lease_superseded"] = True
+        reasons.append(
+            "any reviewer lease scoped to the former head is superseded at the new head"
+        )
+
+    mer_lease = _normalize_sha(former_merger_lease_head_sha)
+    if mer_lease and mer_lease != new:
+        result["merger_lease_superseded"] = True
+        reasons.append(
+            f"merger lease for head {mer_lease} cannot cross to {new}"
+        )
+    else:
+        result["merger_lease_superseded"] = True
+        reasons.append(
+            "any merger lease scoped to the former head is superseded at the new head"
+        )
+
+    prepared = _normalize_sha(prepared_verdict_head_sha)
+    if prepared and prepared != new:
+        result["prepared_verdict_invalidated"] = True
+        reasons.append(
+            f"prepared verdict for head {prepared} cannot authorize the new head {new}"
+        )
+    elif prepared is None:
+        result["prepared_verdict_invalidated"] = True
+        reasons.append(
+            "prepared verdicts associated with the former head are invalidated"
+        )
+
+    result["recommended_next_action"] = ACTION_FRESH_REVIEW_REQUIRED
+    reasons.append(
+        "route the new exact head to independent review "
+        f"({ACTION_FRESH_REVIEW_REQUIRED})"
+    )
+    return result
+
+
+def assess_sequential_queue_step(
+    *,
+    just_merged: bool,
+    master_refreshed: bool,
+    next_pr_number: int | None = None,
+) -> dict[str, Any]:
+    """Controller sequential processing: reassess only after merge + master refresh."""
+    reasons: list[str] = []
+    if just_merged and not master_refreshed:
+        reasons.append(
+            "master must be refreshed after each merge before reassessing the next PR "
+            "(do not update every PR simultaneously)"
+        )
+        return {
+            "reassess_allowed": False,
+            "process_next": False,
+            "next_pr_number": next_pr_number,
+            "reasons": reasons,
+            "success": True,
+        }
+    if not just_merged:
+        reasons.append("no merge completed; sequential step does not advance")
+        return {
+            "reassess_allowed": False,
+            "process_next": False,
+            "next_pr_number": next_pr_number,
+            "reasons": reasons,
+            "success": True,
+        }
+    if next_pr_number:
+        reasons.append(
+            "merge completed and live master refreshed; reassess the next PR "
+            f"#{next_pr_number}"
+        )
+    else:
+        reasons.append(
+            "merge completed and live master refreshed; reassess the next PR"
+        )
+    return {
+        "reassess_allowed": True,
+        "process_next": True,
+        "next_pr_number": next_pr_number,
+        "post_merge_cleanup_handoff": True,
+        "reasons": reasons,
+        "success": True,
+    }
+
+
+def merge_approval_usable_at_head(
+    *,
+    current_head_sha: str | None,
+    approved_head_sha: str | None,
+) -> dict[str, Any]:
+    """Stale approval cannot authorize merge (head-scoped only)."""
+    current = _normalize_sha(current_head_sha)
+    approved = _normalize_sha(approved_head_sha)
+    ok = bool(current and approved and current == approved)
+    reasons: list[str] = []
+    if not ok:
+        reasons.append(
+            f"stale approval: approved head '{approved or '(none)'}' does not "
+            f"match current head '{current or '(none)'}'; cannot authorize merge"
+        )
+    return {
+        "approval_usable": ok,
+        "current_head_sha": current,
+        "approved_head_sha": approved,
+        "reasons": reasons,
+    }
+
+
+def lease_usable_at_head(
+    *,
+    lease_kind: str,
+    lease_head_sha: str | None,
+    current_head_sha: str | None,
+) -> dict[str, Any]:
+    """Reviewer/merger leases cannot cross heads."""
+    current = _normalize_sha(current_head_sha)
+    lease_head = _normalize_sha(lease_head_sha)
+    kind = (lease_kind or "").strip().lower() or "unknown"
+    ok = bool(current and lease_head and current == lease_head)
+    reasons: list[str] = []
+    if not ok:
+        reasons.append(
+            f"stale {kind} lease: lease head '{lease_head or '(none)'}' cannot "
+            f"authorize work at current head '{current or '(none)'}'"
+        )
+    return {
+        "lease_usable": ok,
+        "lease_kind": kind,
+        "lease_head_sha": lease_head,
+        "current_head_sha": current,
+        "reasons": reasons,
+    }

requirements.txt

@@ -31,6 +31,7 @@ python-multipart==0.0.32
 referencing==0.37.0
 rich==15.0.0
 rpds-py==2026.5.1
+sentry-sdk==2.20.0
 shellingham==1.5.4
 sse-starlette==3.4.5
 starlette==1.3.1

reviewer_pr_lease.py

@@ -407,18 +407,194 @@ def record_session_lease(
     if lease_provenance:
         stored["lease_provenance"] = dict(lease_provenance)
     _SESSION_LEASE = stored
+    _persist_session_lease_shadow(stored)
     return dict(_SESSION_LEASE)
 
 
 def clear_session_lease() -> None:
     global _SESSION_LEASE
+    prior = _SESSION_LEASE
     _SESSION_LEASE = None
+    _persist_session_lease_shadow(None, prior=prior)
 
 
 def get_session_lease() -> dict[str, Any] | None:
     return dict(_SESSION_LEASE) if _SESSION_LEASE else None
 
 
+def _persist_session_lease_shadow(
+    lease: dict[str, Any] | None,
+    *,
+    prior: dict[str, Any] | None = None,
+) -> None:
+    """Best-effort durable shadow of the in-session lease (#702).
+
+    A daemon that dies without teardown leaves this record behind, giving a
+    later process provable orphan evidence (owner pid + session id) for the
+    guarded cleanup path. Observability only — mutation gates never read it,
+    and failures here must never block lease operations.
+
+    Shadows are keyed by lease session id (#702 F5): concurrent same-profile
+    sessions each own a distinct record, so one session's heartbeat can never
+    overwrite or misattribute another's crash evidence. A sanctioned clear is
+    the terminal reconciliation of that record's lifecycle.
+    """
+    try:
+        import mcp_session_state as mss
+
+        if lease is None:
+            sid = ((prior or {}).get("session_id") or "").strip() or None
+            if sid:
+                mss.clear_state(
+                    kind=mss.KIND_REVIEWER_SESSION_LEASE, instance_id=sid
+                )
+            # Legacy single-slot record from pre-F5 code: clearing it is safe
+            # because collision-safe writes never target that key again.
+            mss.clear_state(kind=mss.KIND_REVIEWER_SESSION_LEASE)
+            return
+        mss.save_state(
+            kind=mss.KIND_REVIEWER_SESSION_LEASE,
+            instance_id=((lease.get("session_id") or "").strip() or None),
+            payload={
+                "pr_number": lease.get("pr_number"),
+                "session_id": lease.get("session_id"),
+                "comment_id": lease.get("comment_id"),
+                "candidate_head": lease.get("candidate_head"),
+                "worktree": lease.get("worktree"),
+                "phase": lease.get("phase"),
+                "reviewer_identity": lease.get("reviewer_identity"),
+                "profile": lease.get("profile"),
+                "lease_repo": lease.get("repo"),
+                "owner_pid": os.getpid(),
+            },
+        )
+    except Exception:
+        pass
+
+
+def load_session_lease_shadow(
+    session_id: str | None = None,
+) -> dict[str, Any] | None:
+    """Load the durable session-lease shadow left by this profile identity.
+
+    With *session_id*, load that session's collision-safe record (#702 F5),
+    falling back to the legacy single-slot record only when its payload names
+    the same session. Without *session_id*, return the legacy record or the
+    sole surviving instance record — ambiguity (multiple candidates) returns
+    ``None`` because a shadow that cannot be attributed proves nothing.
+    """
+    try:
+        import mcp_session_state as mss
+
+        sid = (session_id or "").strip()
+        if sid:
+            record = mss.load_state(
+                kind=mss.KIND_REVIEWER_SESSION_LEASE, instance_id=sid
+            )
+            if record is not None:
+                return record
+            legacy = mss.load_state(kind=mss.KIND_REVIEWER_SESSION_LEASE)
+            if legacy and (legacy.get("session_id") or "").strip() == sid:
+                return legacy
+            return None
+        legacy = mss.load_state(kind=mss.KIND_REVIEWER_SESSION_LEASE)
+        if legacy is not None:
+            return legacy
+        records = mss.list_states(kind=mss.KIND_REVIEWER_SESSION_LEASE)
+        if len(records) == 1:
+            return records[0]
+        return None
+    except Exception:
+        return None
+
+
+def _pid_alive(pid: int) -> bool:
+    """Probe process existence; fail closed toward 'alive' when unsure."""
+    try:
+        os.kill(int(pid), 0)
+        return True
+    except ProcessLookupError:
+        return False
+    except PermissionError:
+        return True
+    except Exception:
+        return True
+
+
+def assess_crashed_session_lease_orphan(
+    shadow: dict[str, Any] | None,
+    *,
+    lease: dict[str, Any] | None,
+    current_pid: int | None = None,
+    pid_alive: bool | None = None,
+) -> dict[str, Any]:
+    """Derive owner-process evidence for a durable lease from the shadow (#702).
+
+    Returns ``owner_process_alive`` tri-state. Only a provably dead recorded
+    owner pid yields ``False`` (a dead pid cannot still be running the owning
+    daemon); an alive pid is *not* ownership proof and stays ``None`` unless
+    it is this very process. Session ids must match exactly — the shadow of a
+    different lease proves nothing.
+    """
+    reasons: list[str] = []
+    if not shadow or not lease:
+        return {
+            "orphan_evidence": False,
+            "owner_process_alive": None,
+            "reasons": ["no durable session-lease shadow evidence available"],
+        }
+    shadow_sid = (shadow.get("session_id") or "").strip()
+    lease_sid = (lease.get("session_id") or "").strip()
+    if not shadow_sid or not lease_sid or shadow_sid != lease_sid:
+        return {
+            "orphan_evidence": False,
+            "owner_process_alive": None,
+            "reasons": [
+                "session-lease shadow session_id does not match the durable "
+                f"lease (shadow={shadow_sid or None}, lease={lease_sid or None})"
+            ],
+        }
+    owner_pid = shadow.get("owner_pid") or shadow.get("writer_pid")
+    try:
+        owner_pid = int(owner_pid)
+    except (TypeError, ValueError):
+        return {
+            "orphan_evidence": False,
+            "owner_process_alive": None,
+            "reasons": ["session-lease shadow has no usable owner pid (fail closed)"],
+        }
+    this_pid = current_pid if current_pid is not None else os.getpid()
+    if owner_pid == this_pid:
+        return {
+            "orphan_evidence": False,
+            "owner_process_alive": True,
+            "owner_pid": owner_pid,
+            "reasons": ["shadow owner pid is the current process"],
+        }
+    alive = pid_alive if pid_alive is not None else _pid_alive(owner_pid)
+    if alive:
+        reasons.append(
+            f"shadow owner pid {owner_pid} observed alive; PID liveness is not "
+            "ownership proof — owner evidence stays unknown (fail closed)"
+        )
+        return {
+            "orphan_evidence": False,
+            "owner_process_alive": None,
+            "owner_pid": owner_pid,
+            "reasons": reasons,
+        }
+    reasons.append(
+        f"shadow owner pid {owner_pid} is dead: the daemon that recorded the "
+        "matching session lease exited without teardown (crash orphan, #702)"
+    )
+    return {
+        "orphan_evidence": True,
+        "owner_process_alive": False,
+        "owner_pid": owner_pid,
+        "reasons": reasons,
+    }
+
+
 def assess_mutation_lease_gate(
     *,
     pr_number: int,
@@ -557,6 +733,7 @@ _HANDOFF_CLASSIFICATIONS = frozenset({
     "foreign_completed_superseded_head",
     "foreign_expired_superseded_head",
     "orphaned_owner_missing",
+    "orphaned_expired_superseded_head",
     "ambiguous_conflicting_evidence",
     "instructed_lease_missing_with_replacement",
     "worktree_binding_mismatch",
@@ -850,6 +1027,38 @@ def assess_obsolete_reviewer_comment_lease_cleanup(
             classification = "foreign_expired_superseded_head"
         elif head_superseded and has_terminal and not past_expiry:
             classification = "foreign_completed_superseded_head"
+        elif head_superseded and not has_terminal and past_expiry:
+            # Crash-orphan shape (#702): the lease outlived its head with no
+            # formal terminal review and has now passed canonical expiry, so
+            # it no longer gates fresh acquisition. Guarded cleanup of the
+            # ledger marker still demands provable owner-process-exit
+            # evidence and a safe worktree — never inferred.
+            classification = "orphaned_expired_superseded_head"
+            if owner_process_alive is True:
+                fail_closed_reasons.append(
+                    "lease past expiry on superseded head but owner process "
+                    "still alive; cleanup denied"
+                )
+            elif owner_process_alive is None:
+                fail_closed_reasons.append(
+                    "expired superseded-head lease with no terminal review: "
+                    "cleanup requires provable owner-process-exit evidence "
+                    "(owner_process_alive=false); the expired lease no longer "
+                    "gates fresh acquisition"
+                )
+            elif worktree_clean is False:
+                fail_closed_reasons.append(
+                    "owner process absent but worktree dirty; cleanup denied"
+                )
+            elif not (worktree_clean is True or worktree_exists is False):
+                # #702 F3: unknown worktree evidence must fail closed, exactly
+                # like the sibling orphaned_owner_missing gate and the
+                # diagnosis path — cleanup needs affirmative safe evidence.
+                fail_closed_reasons.append(
+                    "owner process absent but worktree evidence is unknown; "
+                    "cleanup requires affirmative safe evidence "
+                    "(worktree_clean=true or worktree_exists=false)"
+                )
         elif head_superseded and not has_terminal:
             classification = "ambiguous_conflicting_evidence"
             fail_closed_reasons.append(
@@ -891,6 +1100,7 @@ def assess_obsolete_reviewer_comment_lease_cleanup(
                 "foreign_expired_superseded_head",
                 "foreign_expired_current_head",
                 "orphaned_owner_missing",
+                "orphaned_expired_superseded_head",
             }:
                 fail_closed_reasons.append(
                     "controller/recovery capability required "
@@ -909,6 +1119,7 @@ def assess_obsolete_reviewer_comment_lease_cleanup(
             "foreign_completed_superseded_head",
             "foreign_expired_superseded_head",
             "orphaned_owner_missing",
+            "orphaned_expired_superseded_head",
             "foreign_expired_current_head",
         }
         # foreign_expired_current_head needs orphan/clean worktree + authority.
@@ -1181,6 +1392,30 @@ def diagnose_reviewer_pr_lease_handoff(
                 "foreign lease pinned to superseded head with completed formal "
                 "review; use guarded obsolete-lease cleanup (do not wait indefinitely)"
             )
+        elif not is_own and head_superseded and not terminal and past_expiry:
+            # Crash-orphan shape after canonical expiry (#702): the expired
+            # marker no longer gates acquisition, so a fresh reviewer may
+            # acquire at the current head. Guarded ledger cleanup becomes
+            # available only with provable owner-exit evidence.
+            classification = "orphaned_expired_superseded_head"
+            if owner_process_alive is False and (
+                worktree_clean is True or worktree_exists is False
+            ):
+                next_action = NEXT_ACTION_CLEANUP_OBSOLETE_LEASE
+                reasons.append(
+                    "expired superseded-head lease with no terminal review and "
+                    "provable owner-process exit (crash orphan, #702); guarded "
+                    "obsolete-lease cleanup eligible"
+                )
+            else:
+                next_action = NEXT_ACTION_ACQUIRE
+                reasons.append(
+                    "lease head superseded and lease past canonical expiry with "
+                    "no formal terminal review (crash orphan, #702); expired "
+                    "marker no longer gates acquisition — acquire a fresh lease "
+                    "at the current head; guarded cleanup needs "
+                    "owner_process_alive=false and a clean/absent worktree"
+                )
         elif not is_own and head_superseded and not terminal:
             classification = "ambiguous_conflicting_evidence"
             next_action = NEXT_ACTION_WAIT

sentry_observability.py

@@ -0,0 +1,535 @@
+"""Optional self-hosted Sentry observability for the Gitea MCP server (#606).
+
+Adds env-var-gated Sentry SDK instrumentation so runtime errors, fail-closed
+workflow blockers, lease/terminal-lock/stale-runtime collisions, and recurring
+watchdog check-ins are visible in a *self-hosted* Sentry at
+``https://sentry.prgs.cc/`` — never Sentry Cloud, and never as the workflow
+source of truth (Gitea stays canonical).
+
+Design constraints (mirror ``gitea_audit`` and the #612 incident bridge):
+
+- **Off by default.** With ``MCP_SENTRY_ENABLED`` false/unset *or* ``SENTRY_DSN``
+  empty, ``init_sentry`` is a no-op and no events are ever sent — existing tool
+  behaviour and API-call patterns are unchanged (acceptance criterion 1).
+- **Fail *open* for observability.** A Sentry outage, a missing ``sentry_sdk``
+  package, or any capture error must never break an MCP tool success path. Every
+  public entry point swallows its own exceptions.
+- **Fail *closed* for redaction.** If a field cannot be proven safe it is dropped
+  rather than sent. Tokens, passwords, keychain IDs, DSNs, private config, raw
+  session-state, full prompt bodies, and full filesystem paths never leave here.
+- **No hard dependency.** ``sentry_sdk`` is imported lazily; the module is fully
+  importable and testable without it installed.
+
+Sentry is observe-only: it must not approve, merge, close, or otherwise mutate
+Gitea workflow state, nor bypass leases, #332, or MCP gates. Alerts may only feed
+the sanctioned Gitea issue/comment path via the #612 incident bridge.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import os
+import re
+from dataclasses import dataclass
+from typing import Any
+
+# Reuse the most comprehensive existing scrubber so redaction stays consistent
+# with the #612 incident bridge (tokens, DSNs, cookies, bearer/basic, keychain
+# ids, session ids, user:pass@host).
+from incident_bridge import redact_text as _redact_text
+
+# Second, complementary scrubber: catches bare ``token <value>`` /
+# ``Bearer <value>`` / ``Basic <value>`` prefixes and raw URLs that the
+# incident-bridge delimiter patterns miss.
+from gitea_audit import _redact_str as _redact_prefixes
+
+# ── Optional SDK (lazy, never a hard dependency) ────────────────────────────
+try:  # pragma: no cover - trivial import guard
+    import sentry_sdk  # type: ignore
+except Exception:  # pragma: no cover - absence is a supported state
+    sentry_sdk = None  # type: ignore
+
+
+# ── Env var names (single source of truth) ──────────────────────────────────
+ENV_ENABLED = "MCP_SENTRY_ENABLED"
+ENV_DSN = "SENTRY_DSN"
+ENV_ENVIRONMENT = "SENTRY_ENVIRONMENT"
+ENV_RELEASE = "SENTRY_RELEASE"
+ENV_TRACES_SAMPLE_RATE = "MCP_SENTRY_TRACES_SAMPLE_RATE"
+ENV_ENABLE_LOGS = "MCP_SENTRY_ENABLE_LOGS"
+
+_TRUTHY = frozenset({"1", "true", "yes", "on"})
+
+REDACTED = "[REDACTED]"
+REDACTED_PATH = "[REDACTED_PATH]"
+
+
+# ── Cron / watchdog monitor slugs (acceptance criterion 6) ──────────────────
+# Stable slugs for the recurring/watchdog jobs #606 wants check-ins for. The
+# slug is the durable monitor identity in Sentry; the wiring call sites pass one
+# of these keys (or an explicit slug) to ``monitor_checkin``.
+MONITOR_SLUGS: dict[str, str] = {
+    "stale_lease_scan": "gitea-mcp-stale-lease-scan",
+    "terminal_lock_scan": "gitea-mcp-terminal-lock-scan",
+    "allocator_health": "gitea-mcp-allocator-health",
+    "namespace_health": "gitea-mcp-namespace-health",
+    "dashboard_freshness": "gitea-mcp-dashboard-freshness",
+    "reconciler_cleanup": "gitea-mcp-reconciler-cleanup",
+}
+
+_CHECKIN_STATUSES = frozenset({"in_progress", "ok", "error"})
+
+
+# ── Tag allowlist (issue "Suggested Sentry tags/context") ───────────────────
+# Only these keys are ever attached as Sentry tags. Anything else is dropped so
+# a caller cannot accidentally leak a sensitive value through a tag.
+ALLOWED_TAG_KEYS = frozenset({
+    "role",
+    "profile",
+    "namespace",
+    "repo",
+    "org",
+    "issue_number",
+    "pr_number",
+    "blocker_type",
+    "workflow_hash",
+    "session_id_hash",  # hash only — never the raw session id
+    "pid",
+    "worktree_category",  # category, never the full sensitive path
+    "lease_comment_id",
+    "expected_head_sha",
+    "current_head_sha",
+    "terminal_lock_state",
+    "capability",
+    "mutation_tool",
+})
+
+# Absolute-path shapes that must never be sent verbatim (macOS/Linux + temp).
+_PATH_RE = re.compile(r"(?:/private)?/(?:Users|home|tmp|var|opt|Volumes)/[^\s\"']*")
+
+# ``extra`` keys whose *full* contents are forbidden by the redaction rules
+# (raw session-state, full prompt/comment bodies, private config blobs, raw
+# headers).
+_FORBIDDEN_EXTRA_KEYS = frozenset({
+    "prompt",
+    "prompt_body",
+    "next_prompt",
+    "body",
+    "raw_body",
+    "session_state",
+    "session_state_contents",
+    "config",
+    "config_contents",
+    "private_config",
+    "headers",
+    "authorization",
+})
+
+
+# ── Configuration ───────────────────────────────────────────────────────────
+@dataclass(frozen=True)
+class SentryConfig:
+    """Immutable snapshot of the Sentry env configuration."""
+
+    enabled: bool = False
+    dsn: str | None = None
+    environment: str = "development"
+    release: str | None = None
+    traces_sample_rate: float = 0.0
+    enable_logs: bool = False
+
+    @property
+    def active(self) -> bool:
+        """True only when the operator both opted in *and* supplied a DSN.
+
+        This is the single gate that keeps the feature off by default: enabling
+        the flag without a DSN (or vice versa) sends nothing.
+        """
+        return bool(self.enabled and self.dsn)
+
+    def safe_summary(self) -> dict[str, Any]:
+        """Operator-facing status with **no** DSN value (only presence)."""
+        return {
+            "enabled": self.enabled,
+            "dsn_present": bool(self.dsn),
+            "environment": self.environment,
+            "release": self.release,
+            "traces_sample_rate": self.traces_sample_rate,
+            "enable_logs": self.enable_logs,
+            "active": self.active,
+        }
+
+
+def _env_bool(name: str, env: dict[str, str]) -> bool:
+    return (env.get(name) or "").strip().lower() in _TRUTHY
+
+
+def _env_float(name: str, default: float, env: dict[str, str]) -> float:
+    raw = (env.get(name) or "").strip()
+    if not raw:
+        return default
+    try:
+        val = float(raw)
+    except (TypeError, ValueError):
+        return default
+    # Clamp to Sentry's valid [0.0, 1.0] sample-rate range.
+    if val < 0.0:
+        return 0.0
+    if val > 1.0:
+        return 1.0
+    return val
+
+
+def load_config(env: dict[str, str] | None = None) -> SentryConfig:
+    """Build a :class:`SentryConfig` from the environment (read at call time)."""
+    env = dict(os.environ if env is None else env)
+    dsn = (env.get(ENV_DSN) or "").strip() or None
+    return SentryConfig(
+        enabled=_env_bool(ENV_ENABLED, env),
+        dsn=dsn,
+        environment=(env.get(ENV_ENVIRONMENT) or "").strip() or "development",
+        release=(env.get(ENV_RELEASE) or "").strip() or None,
+        traces_sample_rate=_env_float(ENV_TRACES_SAMPLE_RATE, 0.0, env),
+        enable_logs=_env_bool(ENV_ENABLE_LOGS, env),
+    )
+
+
+def sdk_available() -> bool:
+    """True when the optional ``sentry_sdk`` package is importable."""
+    return sentry_sdk is not None
+
+
+# ── Redaction (fail closed) ─────────────────────────────────────────────────
+def sanitize_path(value: Any) -> str:
+    """Reduce a filesystem path to a non-sensitive *category* token.
+
+    Full local paths must never be sent. We keep only a coarse worktree
+    category derived from the path shape (author/reviewer/merger/reconciler/
+    branches/root/other).
+    """
+    text = "" if value is None else str(value)
+    low = text.lower()
+    if not text:
+        return "unknown"
+    # Order matters: more specific role markers before the generic "branches".
+    if "reconcile" in low:
+        return "reconciler"
+    if "review" in low:
+        return "reviewer"
+    if "merge" in low or "merger" in low:
+        return "merger"
+    if "author" in low or re.search(r"/branches/(?:feat|fix|docs|chore|issue)", low):
+        return "author"
+    if "/branches/" in low:
+        return "branches"
+    if low.rstrip("/").endswith("gitea-tools"):
+        return "root"
+    return "other"
+
+
+def redact_value(value: Any) -> Any:
+    """Recursively redact a JSON-able value: secret text, absolute paths, and
+    known-sensitive dict keys are removed. Fail closed — any error drops the
+    value entirely rather than risk leaking it."""
+    try:
+        if isinstance(value, dict):
+            out: dict[str, Any] = {}
+            for k, v in value.items():
+                key = str(k)
+                low = key.lower()
+                if low in _FORBIDDEN_EXTRA_KEYS or any(
+                    s in low
+                    for s in ("token", "secret", "password", "cookie", "auth", "dsn", "keychain")
+                ):
+                    out[key] = REDACTED
+                    continue
+                out[key] = redact_value(v)
+            return out
+        if isinstance(value, (list, tuple)):
+            return [redact_value(v) for v in value]
+        if isinstance(value, str):
+            scrubbed = _redact_text(value)
+            scrubbed = _redact_prefixes(scrubbed)
+            scrubbed = _PATH_RE.sub(REDACTED_PATH, scrubbed)
+            return scrubbed
+        return value
+    except Exception:
+        return REDACTED
+
+
+def hash_session_id(session_id: Any) -> str:
+    """Short, stable, non-reversible fingerprint of a session id."""
+    digest = hashlib.sha256(str(session_id).encode("utf-8", "replace")).hexdigest()
+    return digest[:12]
+
+
+def build_tags(**kwargs: Any) -> dict[str, str]:
+    """Return a scrubbed, allowlisted tag dict.
+
+    ``session_id`` is accepted but only ever surfaced as ``session_id_hash``.
+    ``worktree_path`` collapses to ``worktree_category``. Any non-allowlisted
+    key, or a value that still contains redacted material after scrubbing, is
+    dropped.
+    """
+    raw: dict[str, Any] = dict(kwargs)
+
+    # Hash the session id — never emit it raw.
+    session_id = raw.pop("session_id", None)
+    if session_id and "session_id_hash" not in raw:
+        raw["session_id_hash"] = hash_session_id(session_id)
+
+    # A full worktree path collapses to a category tag.
+    wt = raw.pop("worktree_path", None)
+    if wt and "worktree_category" not in raw:
+        raw["worktree_category"] = sanitize_path(wt)
+
+    out: dict[str, str] = {}
+    for key, val in raw.items():
+        if key not in ALLOWED_TAG_KEYS:
+            continue
+        if val is None:
+            continue
+        scrubbed = redact_value(val)
+        text = str(scrubbed)
+        if not text or REDACTED in text or REDACTED_PATH in text:
+            continue
+        if len(text) > 200:
+            text = text[:200] + "…"
+        out[key] = text
+    return out
+
+
+def scrub_event(event: Any, hint: Any = None) -> dict[str, Any] | None:
+    """Sentry ``before_send`` / ``before_send_log`` hook.
+
+    Recursively redacts the outgoing event. On *any* failure it returns ``None``
+    so the event is dropped rather than sent unscrubbed (fail closed for
+    redaction).
+    """
+    try:
+        if not isinstance(event, dict):
+            return None
+        scrubbed = redact_value(event)
+        # Drop server_name if it leaked a hostname/path; PID is kept via tags.
+        scrubbed.pop("server_name", None)
+        return scrubbed
+    except Exception:
+        return None
+
+
+# ── Event builders (pure, independently testable) ───────────────────────────
+def build_blocker_event(
+    blocker_type: str,
+    *,
+    message: str | None = None,
+    next_action: str | None = None,
+    level: str = "warning",
+    tags: dict[str, Any] | None = None,
+    extra: dict[str, Any] | None = None,
+) -> dict[str, Any]:
+    """Build a redacted, structured Sentry event for a workflow blocker.
+
+    ``next_action`` maps the issue's "canonical next action when available"
+    requirement (acceptance criterion 7).
+    """
+    merged_tags = dict(tags or {})
+    merged_tags.setdefault("blocker_type", blocker_type)
+    safe_tags = build_tags(**merged_tags)
+
+    safe_extra = redact_value(dict(extra or {}))
+    if next_action:
+        # A short canonical next action is allowed (it is not a full prompt).
+        safe_extra["canonical_next_action"] = redact_value(str(next_action)[:500])
+
+    event: dict[str, Any] = {
+        "message": redact_value(message or blocker_type),
+        "level": level if level in ("debug", "info", "warning", "error", "fatal") else "warning",
+        "logger": "gitea-mcp.workflow",
+        "tags": safe_tags,
+        "extra": safe_extra,
+        "fingerprint": ["workflow-blocker", blocker_type],
+    }
+    return event
+
+
+def build_checkin_payload(
+    monitor: str,
+    status: str,
+    *,
+    check_in_id: str | None = None,
+    duration: float | None = None,
+) -> dict[str, Any]:
+    """Build a Sentry cron check-in payload for one of :data:`MONITOR_SLUGS`.
+
+    ``monitor`` may be a registry key (e.g. ``"stale_lease_scan"``) or an
+    explicit slug. Raises ``ValueError`` on an unknown status so callers cannot
+    silently send a malformed check-in.
+    """
+    if status not in _CHECKIN_STATUSES:
+        raise ValueError(
+            f"invalid check-in status {status!r}; expected one of {sorted(_CHECKIN_STATUSES)}"
+        )
+    slug = MONITOR_SLUGS.get(monitor, monitor)
+    payload: dict[str, Any] = {"monitor_slug": slug, "status": status}
+    if check_in_id:
+        payload["check_in_id"] = str(check_in_id)
+    if duration is not None:
+        try:
+            payload["duration"] = float(duration)
+        except (TypeError, ValueError):
+            pass
+    return payload
+
+
+# ── Runtime init + capture (fail open) ──────────────────────────────────────
+_STATE: dict[str, Any] = {"initialized": False, "config": None}
+
+
+def is_initialized() -> bool:
+    return bool(_STATE.get("initialized"))
+
+
+def active_config() -> SentryConfig | None:
+    return _STATE.get("config")
+
+
+def reset_for_tests() -> None:
+    """Clear module init state. Test-only helper (never called in production)."""
+    _STATE["initialized"] = False
+    _STATE["config"] = None
+
+
+def init_sentry(config: SentryConfig | None = None) -> dict[str, Any]:
+    """Initialise the Sentry SDK if (and only if) enabled + DSN + SDK present.
+
+    Idempotent and never raises. Returns an operator-safe status dict (no DSN
+    value). Behaviour is unchanged when the feature is off.
+    """
+    cfg = config or load_config()
+    status: dict[str, Any] = {"initialized": False, **cfg.safe_summary()}
+    try:
+        if not cfg.active:
+            status["reason"] = "disabled (MCP_SENTRY_ENABLED false or SENTRY_DSN empty)"
+            _STATE["config"] = cfg
+            return status
+        if not sdk_available():
+            status["reason"] = "sentry_sdk not installed"
+            _STATE["config"] = cfg
+            return status
+
+        init_kwargs: dict[str, Any] = {
+            "dsn": cfg.dsn,
+            "environment": cfg.environment,
+            "release": cfg.release,
+            "traces_sample_rate": cfg.traces_sample_rate,
+            "before_send": scrub_event,
+            "send_default_pii": False,
+        }
+        if cfg.enable_logs:
+            # sentry-sdk 2.x captures Python logs as structured logs when the
+            # experimental logs feature is enabled; scrub those too.
+            init_kwargs["_experiments"] = {
+                "enable_logs": True,
+                "before_send_log": scrub_event,
+            }
+        sentry_sdk.init(**init_kwargs)  # type: ignore[union-attr]
+        _STATE["initialized"] = True
+        _STATE["config"] = cfg
+        status["initialized"] = True
+        status["reason"] = "sentry initialised"
+    except Exception as exc:  # fail open: observability must not block startup
+        status["reason"] = f"init failed (ignored): {type(exc).__name__}"
+        _STATE["initialized"] = False
+    return status
+
+
+def _set_scope_tags(scope: Any, tags: dict[str, str]) -> None:
+    for key, val in tags.items():
+        try:
+            scope.set_tag(key, val)
+        except Exception:
+            pass
+
+
+def capture_workflow_blocker(
+    blocker_type: str,
+    *,
+    message: str | None = None,
+    next_action: str | None = None,
+    level: str = "warning",
+    tags: dict[str, Any] | None = None,
+    extra: dict[str, Any] | None = None,
+) -> dict[str, Any]:
+    """Capture a fail-closed workflow blocker as a structured Sentry event.
+
+    Always returns the redacted event dict (so callers/tests can inspect it),
+    and sends it to Sentry only when initialised. Fail open.
+    """
+    event = build_blocker_event(
+        blocker_type,
+        message=message,
+        next_action=next_action,
+        level=level,
+        tags=tags,
+        extra=extra,
+    )
+    try:
+        if is_initialized() and sdk_available():
+            sentry_sdk.capture_event(event)  # type: ignore[union-attr]
+    except Exception:
+        pass
+    return event
+
+
+def capture_exception(
+    exc: BaseException,
+    *,
+    tags: dict[str, Any] | None = None,
+    extra: dict[str, Any] | None = None,
+) -> bool:
+    """Capture a runtime exception with scrubbed tags. Fail open.
+
+    Returns True only when the event was handed to an initialised SDK.
+    """
+    try:
+        if not (is_initialized() and sdk_available()):
+            return False
+        safe_tags = build_tags(**(tags or {}))
+        safe_extra = redact_value(dict(extra or {}))
+        with sentry_sdk.push_scope() as scope:  # type: ignore[union-attr]
+            _set_scope_tags(scope, safe_tags)
+            for key, val in safe_extra.items():
+                try:
+                    scope.set_extra(key, val)
+                except Exception:
+                    pass
+            sentry_sdk.capture_exception(exc)  # type: ignore[union-attr]
+        return True
+    except Exception:
+        return False
+
+
+def monitor_checkin(
+    monitor: str,
+    status: str,
+    *,
+    check_in_id: str | None = None,
+    duration: float | None = None,
+) -> dict[str, Any] | None:
+    """Send a Sentry cron check-in for a watchdog job. Fail open.
+
+    Returns the payload (for inspection/tests), or ``None`` if the status was
+    invalid. Only transmits when initialised.
+    """
+    try:
+        payload = build_checkin_payload(
+            monitor, status, check_in_id=check_in_id, duration=duration
+        )
+    except ValueError:
+        return None
+    try:
+        if is_initialized() and sdk_available() and hasattr(sentry_sdk, "capture_checkin"):
+            sentry_sdk.capture_checkin(**payload)  # type: ignore[union-attr]
+    except Exception:
+        pass
+    return payload

skills/llm-project-workflow/templates/merge-pr.md

@@ -33,22 +33,34 @@ Steps:
    *If the current identity does not match the required role (or is the PR author), STOP. Relaunch/switch to the correct profile first.*
 2. Verify authenticated identity + active profile.
 3. Confirm PR #<pr>: author (not you), state open, mergeable, review approved. Check if PR body uses `Closes #N` or `Fixes #N`; if it uses `Implements #N` or `Refs #N`, manual closing will be needed in step 29.
-4. Capability evidence (#179): cite the exact gitea_resolve_task_capability
+4. **PR sync assess (required):** call `gitea_assess_pr_sync_status` with
+   explicit `remote`/`org`/`repo`. Route on `recommended_next_action`:
+   - `merge_now` → continue merger path (do **not** update the branch)
+   - `update_branch_by_merge` → STOP; hand off to **author** for
+     `gitea_update_pr_branch_by_merge` (pins expected PR head + base head)
+   - `author_conflict_remediation` → STOP; author worktree conflict fix
+   - `fresh_review_required` → STOP; independent re-review at current head
+   - `blocked` → STOP and diagnose
+   Never merge on a former-head approval after update/remediation.
+5. Capability evidence (#179): cite the exact gitea_resolve_task_capability
    output (or runtime context) proving merge_pr is allowed — a bare
    "capability checks passed" claim is downgraded.
-5. Final live-state recheck (#179), immediately before the merge mutation —
+6. Final live-state recheck (#179), immediately before the merge mutation —
    re-read the live PR and prove:
    - PR still open
    - live head SHA still equals the pinned/reviewed head SHA
    - base branch unchanged
    - no undismissed REQUEST_CHANGES / blocking review state remains
    If any recheck fails → STOP, re-pin, re-validate.
-6. If any gate fails → STOP and report.
-7. Merge with explicit confirmation (e.g. confirmation="MERGE PR <pr>"),
+7. If any gate fails → STOP and report.
+8. Merge with explicit confirmation (e.g. confirmation="MERGE PR <pr>"),
    pinning the reviewed head SHA (expected_head_sha) and, where supported,
    the changed-file set.
-8. Confirm remote master now contains the merge commit (or the expected changes if squash merged).
+9. Confirm remote master now contains the merge commit (or the expected changes if squash merged).
    *Note: Gitea PR "closed" state is NOT equivalent to "merged". Do not assume a closed PR succeeded without verifying the actual landed changes.*
+10. **Sequential queue:** after merge, refresh live `master`, run post-merge
+    cleanup handoff, then reassess the **next** PR (do not batch-update all
+    open PRs).
 
 Post-merge cleanup (#517): merger sessions must NOT perform ad hoc cleanup.
 - Record merge mutations separately from cleanup mutations in the controller handoff.

skills/llm-project-workflow/workflows/review-merge-pr.md

@@ -950,9 +950,53 @@ Final reports must state:
 * final live head SHA before merge
 * whether any push occurred during validation
 
+## 26D. PR synchronization and conflict-remediation lifecycle
+
+**Do not treat “approved” as the final readiness state.** For every approved
+open PR, call `gitea_assess_pr_sync_status` (native MCP only) and route by
+`recommended_next_action`:
+
+| Action | Meaning | Next role / tools |
+|--------|---------|-------------------|
+| `merge_now` | Valid approval at exact current head; conflict-free; update not required; checks ok | Merger: sanctioned merge workflow only — **do not** update the branch |
+| `update_branch_by_merge` | Behind live base; protection requires current base; Gitea can merge base without conflicts | **Author only:** `gitea_update_pr_branch_by_merge` with pinned `expected_pr_head_sha` + `expected_base_head_sha` |
+| `author_conflict_remediation` | Conflicts / not auto-updatable | **Author only:** existing `branches/` worktree, issue lock, merge master, resolve, test, push; never force-push/rebase |
+| `fresh_review_required` | Head changed after approval, or update/remediation produced a new head | Independent reviewer at the **new exact head**; old approvals/leases/verdicts are void |
+| `blocked` | Other gate (checks, incomplete facts, closed PR, …) | Diagnose; do not merge or update |
+
+### Hard rules
+
+* Pin both expected PR head and expected base head for every update. Either
+  race → fail closed with no partial mutation.
+* Never rebase or force-push author branches.
+* Never update an author branch from a reviewer or merger profile.
+* Never preserve approval, reviewer lease, merger lease, or prepared verdict
+  across a head change.
+* Process PRs **sequentially**: synchronize/remediate one → review new head →
+  merge → refresh live `master` → reassess the next PR. Do not update every
+  open PR at once (each merge restales the rest).
+* Conflict remediation only in the existing dedicated issue/PR worktree under
+  `branches/`. Do not delete that worktree until the updated PR is merged and
+  cleanup eligibility is proven.
+* Post-merge: canonical cleanup handoff to reconciler (section 28).
+
+### After `gitea_update_pr_branch_by_merge` succeeds
+
+1. Treat former-head approval as invalidated.
+2. Release/supersede obsolete reviewer and merger leases.
+3. Route `recommended_next_action=fresh_review_required` at the new head.
+4. Independent re-review, then merger lease/adopt + merge at the new head only.
+
+All Gitea reads/mutations use native MCP. Never substitute direct API, curl,
+tea/gh, Web UI mutation by an LLM, database changes, raw Git push, or token
+access.
+
 ## 27. Merge rules
 
-Before merge, rerun fresh live checks:
+Before merge, call `gitea_assess_pr_sync_status` when the PR is approved/open
+and may be behind base. Only proceed with merge when
+`recommended_next_action` is `merge_now` and approval remains at the current
+head. Then rerun fresh live checks:
 
 * whoami
 * active profile/runtime

stale_binding_recovery.py

@@ -0,0 +1,257 @@
+"""Sanctioned recovery for stale env-bound task workspace bindings (#702).
+
+When the MCP daemon dies from an unhandled exception it never runs teardown,
+and the auto-reconnected daemon inherits ``GITEA_ACTIVE_WORKTREE`` from its
+parent environment. That inherited value can permanently bind the fresh
+session to a prior task's worktree (the PR #701 / ``review-pr-654`` incident).
+
+This module classifies the active env binding against live session evidence
+and produces a fail-closed recovery plan. Only two situations permit the
+daemon to clear the binding on its own:
+
+- the bound path no longer exists on disk (``provably_stale_missing_path``)
+- the binding was inherited at daemon boot and a *sanctioned* in-session
+  reviewer/merger lease later bound a different worktree
+  (``superseded_by_session_lease``)
+
+Everything else is surfaced (``unverified_inherited``) and left for the
+managed reconnect path — never cleared silently, never rebound to a guessed
+worktree.
+"""
+
+from __future__ import annotations
+
+import os
+from typing import Any
+
+ACTIVE_WORKTREE_ENV = "GITEA_ACTIVE_WORKTREE"
+
+CLASSIFICATION_UNBOUND = "unbound"
+CLASSIFICATION_CORROBORATED = "corroborated"
+CLASSIFICATION_UNVERIFIED_INHERITED = "unverified_inherited"
+CLASSIFICATION_PROVABLY_STALE_MISSING_PATH = "provably_stale_missing_path"
+CLASSIFICATION_SUPERSEDED_BY_SESSION_LEASE = "superseded_by_session_lease"
+
+CLASSIFICATIONS = frozenset({
+    CLASSIFICATION_UNBOUND,
+    CLASSIFICATION_CORROBORATED,
+    CLASSIFICATION_UNVERIFIED_INHERITED,
+    CLASSIFICATION_PROVABLY_STALE_MISSING_PATH,
+    CLASSIFICATION_SUPERSEDED_BY_SESSION_LEASE,
+})
+
+# Roles whose task workspace is owned by a lease lifecycle, so an inherited
+# generic binding without a corroborating lease is suspect.
+LEASE_BOUND_ROLES = frozenset({"reviewer", "merger"})
+
+RECOVERY_ACTION_NONE = "none"
+RECOVERY_ACTION_CLEAR_ENV = "clear_env"
+
+
+def _norm(path: str | None) -> str:
+    text = (path or "").strip()
+    if not text:
+        return ""
+    return os.path.realpath(os.path.abspath(text))
+
+
+def snapshot_boot_bindings(
+    env: dict[str, str] | os._Environ | None = None,
+) -> dict[str, Any]:
+    """Capture worktree env bindings present when the daemon booted.
+
+    A value present in this snapshot was inherited from the parent
+    environment, not established by any sanctioned tool in this daemon's
+    lifetime.
+    """
+    env_map = env if env is not None else os.environ
+    return {
+        "active_worktree": (env_map.get(ACTIVE_WORKTREE_ENV) or "").strip() or None,
+    }
+
+
+def classify_active_worktree_binding(
+    *,
+    active_value: str | None,
+    role_env_value: str | None = None,
+    session_lease_worktree: str | None = None,
+    boot_inherited: bool,
+    path_exists: bool | None,
+    role_kind: str | None = None,
+) -> dict[str, Any]:
+    """Classify the GITEA_ACTIVE_WORKTREE binding against live evidence.
+
+    Fail closed: unknown evidence never yields a clear-eligible class.
+    """
+    reasons: list[str] = []
+    active = _norm(active_value)
+    if not active:
+        return {
+            "classification": CLASSIFICATION_UNBOUND,
+            "clear_eligible": False,
+            "active_worktree": None,
+            "reasons": ["no GITEA_ACTIVE_WORKTREE binding present"],
+        }
+
+    role = (role_kind or "").strip().lower()
+    role_env = _norm(role_env_value)
+    lease_wt = _norm(session_lease_worktree)
+
+    result: dict[str, Any] = {
+        "active_worktree": active,
+        "boot_inherited": bool(boot_inherited),
+        "role_kind": role or None,
+        "session_lease_worktree": lease_wt or None,
+        "role_env_worktree": role_env or None,
+    }
+
+    if path_exists is False:
+        reasons.append(
+            f"bound worktree '{active}' no longer exists on disk; the binding "
+            "cannot name a valid task workspace"
+        )
+        result.update({
+            "classification": CLASSIFICATION_PROVABLY_STALE_MISSING_PATH,
+            "clear_eligible": True,
+            "reasons": reasons,
+        })
+        return result
+
+    if lease_wt and lease_wt == active:
+        reasons.append("binding matches the sanctioned in-session lease worktree")
+        result.update({
+            "classification": CLASSIFICATION_CORROBORATED,
+            "clear_eligible": False,
+            "reasons": reasons,
+        })
+        return result
+
+    if lease_wt and lease_wt != active and boot_inherited:
+        reasons.append(
+            "boot-inherited binding disagrees with the sanctioned in-session "
+            f"lease worktree '{lease_wt}'; the lease is live authority"
+        )
+        result.update({
+            "classification": CLASSIFICATION_SUPERSEDED_BY_SESSION_LEASE,
+            "clear_eligible": True,
+            "reasons": reasons,
+        })
+        return result
+
+    if lease_wt and lease_wt != active and not boot_inherited:
+        # Set after boot by tooling; disagreement is surfaced, never auto-cleared.
+        reasons.append(
+            "post-boot binding disagrees with the in-session lease worktree; "
+            "surfacing only (fail closed, no automatic clear)"
+        )
+        result.update({
+            "classification": CLASSIFICATION_UNVERIFIED_INHERITED,
+            "clear_eligible": False,
+            "reasons": reasons,
+        })
+        return result
+
+    if role_env and role_env == active:
+        reasons.append("binding matches the role-specific worktree env binding")
+        result.update({
+            "classification": CLASSIFICATION_CORROBORATED,
+            "clear_eligible": False,
+            "reasons": reasons,
+        })
+        return result
+
+    if boot_inherited and role in LEASE_BOUND_ROLES and not lease_wt:
+        reasons.append(
+            "binding was inherited at daemon boot, no sanctioned in-session "
+            f"lease corroborates it, and the {role} role binds workspaces "
+            "through the lease lifecycle; treat as unverified until a fresh "
+            "lease or managed reconnect re-establishes the workspace"
+        )
+        result.update({
+            "classification": CLASSIFICATION_UNVERIFIED_INHERITED,
+            "clear_eligible": False,
+            "reasons": reasons,
+        })
+        return result
+
+    reasons.append("no contradicting evidence; binding treated as corroborated")
+    result.update({
+        "classification": CLASSIFICATION_CORROBORATED,
+        "clear_eligible": False,
+        "reasons": reasons,
+    })
+    return result
+
+
+def plan_recovery(classification_result: dict[str, Any]) -> dict[str, Any]:
+    """Turn a classification into an explicit, fail-closed recovery plan."""
+    classification = classification_result.get("classification")
+    clear_eligible = bool(classification_result.get("clear_eligible"))
+    reasons = list(classification_result.get("reasons") or [])
+
+    if classification not in CLASSIFICATIONS:
+        return {
+            "action": RECOVERY_ACTION_NONE,
+            "clear_allowed": False,
+            "classification": classification,
+            "reasons": reasons + ["unknown classification (fail closed)"],
+        }
+
+    if clear_eligible and classification in {
+        CLASSIFICATION_PROVABLY_STALE_MISSING_PATH,
+        CLASSIFICATION_SUPERSEDED_BY_SESSION_LEASE,
+    }:
+        return {
+            "action": RECOVERY_ACTION_CLEAR_ENV,
+            "clear_allowed": True,
+            "classification": classification,
+            "active_worktree": classification_result.get("active_worktree"),
+            "reasons": reasons,
+        }
+
+    exact_next_action = None
+    if classification == CLASSIFICATION_UNVERIFIED_INHERITED:
+        exact_next_action = (
+            "managed recovery required: reconnect the MCP namespace through "
+            "the managed client configuration (or start a fresh session) so "
+            "the daemon boots without the inherited GITEA_ACTIVE_WORKTREE, or "
+            "corroborate the binding by acquiring a lease for that worktree; "
+            "do not clear or rewrite the environment manually"
+        )
+    return {
+        "action": RECOVERY_ACTION_NONE,
+        "clear_allowed": False,
+        "classification": classification,
+        "active_worktree": classification_result.get("active_worktree"),
+        "reasons": reasons,
+        **({"exact_next_action": exact_next_action} if exact_next_action else {}),
+    }
+
+
+def apply_recovery(
+    plan: dict[str, Any],
+    env: dict[str, str] | os._Environ | None = None,
+) -> dict[str, Any]:
+    """Apply a clear-eligible plan by removing the stale env binding.
+
+    This is the sanctioned in-daemon mechanism (#702 AC2); callers must
+    persist the returned record for audit. Refuses anything but an explicit
+    ``clear_env`` plan.
+    """
+    env_map = env if env is not None else os.environ
+    if not plan.get("clear_allowed") or plan.get("action") != RECOVERY_ACTION_CLEAR_ENV:
+        return {
+            "performed": False,
+            "action": RECOVERY_ACTION_NONE,
+            "reasons": ["recovery plan does not authorize clearing (fail closed)"],
+        }
+    cleared_value = (env_map.get(ACTIVE_WORKTREE_ENV) or "").strip() or None
+    env_map.pop(ACTIVE_WORKTREE_ENV, None)
+    return {
+        "performed": True,
+        "action": RECOVERY_ACTION_CLEAR_ENV,
+        "cleared_env": ACTIVE_WORKTREE_ENV,
+        "cleared_value": cleared_value,
+        "classification": plan.get("classification"),
+        "reasons": list(plan.get("reasons") or []),
+    }

stale_review_decision_lock.py

@@ -1,4 +1,4 @@
-"""Stale / moot #332 review-decision lock detection and cleanup policy (#594/#620).
+"""Stale / moot #332 review-decision lock detection and cleanup policy (#594/#620/#693).
 
 #332 correctly hard-stops a reviewer session after a terminal live review
 mutation. After #559 those locks are durable on disk, so they can outlive the
@@ -12,6 +12,11 @@ on head B of the **same open PR**. Same-head duplicates remain fail-closed.
 Open-PR lock *cleanup* remains forbidden unless the PR is truly merged/closed
 (#594); new-head re-review is allowed without deleting the durable lock.
 
+#693 scopes the terminal boundary by **PR number**: a terminal on open PR A
+must not block a fresh formal decision on open PR B on the same durable
+profile lock (cross-PR isolation). Correction authorization is scoped to the
+named prior review's PR/head and cannot unlock a different PR.
+
 This module is pure policy (no Gitea I/O). The MCP tool
 ``gitea_cleanup_stale_review_decision_lock`` fetches live PR state, calls these
 helpers, and only then clears durable state when cleanup is allowed.
@@ -80,17 +85,45 @@ def last_terminal_mutation(lock: dict | None) -> dict | None:
     return terminals[-1] if terminals else None
 
 
+def correction_applies(
+    lock: dict | None,
+    *,
+    pr_number: int | None = None,
+    expected_head_sha: str | None = None,
+) -> bool:
+    """True when operator correction is active and scoped to the target (#693).
+
+    Global ``correction_authorized`` alone is not enough: correction must name
+    the same PR (and head when recorded) as the decision being re-opened.
+    Missing scope fields on legacy locks fail closed (do not apply).
+    """
+    if not lock or not lock.get("correction_authorized"):
+        return False
+    corr_pr = lock.get("correction_pr_number")
+    if corr_pr is None:
+        # Legacy unscoped correction — refuse as generic unlock (#693).
+        return False
+    if pr_number is not None and int(corr_pr) != int(pr_number):
+        return False
+    corr_head = normalize_head_sha(lock.get("correction_head_sha"))
+    target = normalize_head_sha(expected_head_sha)
+    if corr_head and target and not heads_equal(corr_head, target):
+        return False
+    return True
+
+
 def prior_live_mutations_block_boundary(
     lock: dict | None,
     *,
     pr_number: int,
     expected_head_sha: str | None,
 ) -> bool:
-    """True when prior live mutations block a new decision for PR+head (#620).
+    """True when prior live mutations block a new decision for PR+head (#620/#693).
 
     Historical terminals on a **different head of the same PR** do not block.
-    Any prior mutation on another PR, the same head, or without a comparable
-    head SHA fails closed (blocks).
+    Terminals on a **different PR** do not block (#693 cross-PR isolation).
+    Same PR + same head (or same PR without a comparable head SHA) blocks
+    unless a scoped correction applies.
 
     Head resolution uses ``mutation_head_sha(m, lock)`` so pre-#620 ledgers
     that only stored ``ready_expected_head_sha`` still compare correctly
@@ -98,7 +131,9 @@ def prior_live_mutations_block_boundary(
     """
     if not lock:
         return False
-    if lock.get("correction_authorized"):
+    if correction_applies(
+        lock, pr_number=pr_number, expected_head_sha=expected_head_sha
+    ):
         return False
     prior = list(lock.get("live_mutations") or [])
     if not prior:
@@ -108,10 +143,14 @@ def prior_live_mutations_block_boundary(
         if not isinstance(m, dict):
             return True
         m_pr = m.get("pr_number")
+        if m_pr is None:
+            return True  # fail closed — unscoped mutation
+        if int(m_pr) != int(pr_number):
+            # #693: foreign-PR history on the shared durable lock does not block.
+            continue
         m_head = mutation_head_sha(m, lock)
         if (
-            m_pr == pr_number
-            and target
+            target
             and m_head
             and not heads_equal(m_head, target)
         ):
@@ -128,23 +167,34 @@ def terminal_boundary_allows_fresh_decision(
     expected_head_sha: str | None,
     operation: str,
 ) -> bool:
-    """Whether #332 hard-stop should yield for a new PR+head boundary (#620).
+    """Whether #332 hard-stop should yield for a new PR+head boundary (#620/#693).
 
-    Only for ``mark_ready`` / ``review`` / ``resume`` on the **same PR** when
-    the requested head differs from the last terminal's head. Merge is never
-    reopened by head change (approved head merge path is separate).
+    For ``mark_ready`` / ``review`` / ``resume``:
+
+    * **same PR, different head** — allowed (#620)
+    * **different PR than last terminal** — allowed (#693 cross-PR isolation)
+    * **same PR, same head** — not allowed (unless scoped correction)
+
+    Merge is never reopened by head change or cross-PR isolation (approved
+    head merge path is separate).
     """
     if operation not in ("mark_ready", "review", "resume"):
         return False
     if not lock:
         return False
-    if lock.get("correction_authorized"):
+    if correction_applies(
+        lock, pr_number=pr_number, expected_head_sha=expected_head_sha
+    ):
         return True
     last = last_terminal_mutation(lock)
     if last is None:
         return True
-    if last.get("pr_number") != pr_number:
+    last_pr = last.get("pr_number")
+    if last_pr is None:
         return False
+    if int(last_pr) != int(pr_number):
+        # #693: last terminal belongs to another PR — do not hard-stop this PR.
+        return True
     locked_head = mutation_head_sha(last, lock)
     target = normalize_head_sha(expected_head_sha)
     if not locked_head or not target:
@@ -440,6 +490,400 @@ def format_cleanup_audit_comment(audit: dict[str, Any]) -> str:
     return "\n".join(lines)
 
 
+# --- #693 classification / recovery / correction audit -----------------------
+
+# Deterministic classification labels for diagnostics and recovery routing.
+CLASS_NO_LOCK = "no_lock"
+CLASS_READY_FOR_TARGET = "ready_for_target"
+CLASS_IN_PROGRESS_SAME_HEAD = "in_progress_same_head"
+CLASS_TERMINAL_ACTIVE_SAME_HEAD = "terminal_active_same_head"
+CLASS_STALE_SUPERSEDED_HEAD = "stale_superseded_head"
+CLASS_FOREIGN_PR_TERMINAL = "foreign_pr_terminal"
+CLASS_MOOT_MERGED_CLOSED = "moot_merged_closed"
+CLASS_CORRECTION_ACTIVE = "correction_active"
+CLASS_SESSION_OR_PROFILE_MISMATCH = "session_or_profile_mismatch"
+CLASS_AMBIGUOUS = "ambiguous"
+
+
+def classify_review_decision_lock(
+    lock: dict | None,
+    *,
+    target_pr_number: int | None = None,
+    target_head_sha: str | None = None,
+    pr_live: dict | None = None,
+    pr_lookup_error: str | None = None,
+    active_profile_identity: str | None = None,
+    session_blockers: list[str] | None = None,
+) -> dict[str, Any]:
+    """Deterministic review-decision-lock classification (#693).
+
+    Returns classification, exact next_action, owner/evidence fields, and
+    whether mark_final / cleanup / correction are appropriate.
+    """
+    summary = lock_summary(lock)
+    target_head = normalize_head_sha(target_head_sha)
+    result: dict[str, Any] = {
+        "classification": CLASS_NO_LOCK,
+        "has_lock": lock is not None,
+        "lock_summary": summary,
+        "target_pr_number": target_pr_number,
+        "target_head_sha": target_head,
+        "last_terminal_pr": None,
+        "last_terminal_action": None,
+        "locked_head_sha": None,
+        "owner_profile_identity": (summary or {}).get("profile_identity"),
+        "session_profile": (summary or {}).get("session_profile"),
+        "updated_at": (summary or {}).get("updated_at"),
+        "correction_authorized": bool((lock or {}).get("correction_authorized")),
+        "correction_pr_number": (lock or {}).get("correction_pr_number"),
+        "correction_head_sha": normalize_head_sha(
+            (lock or {}).get("correction_head_sha")
+        ),
+        "mark_final_allowed": True,
+        "cleanup_allowed": False,
+        "correction_eligible": False,
+        "next_action": "gitea_resolve_task_capability(task='review_pr') then mark_final",
+        "reasons": [],
+        "evidence": {},
+    }
+
+    if lock is None:
+        result["reasons"].append("no review decision lock present")
+        return result
+
+    session_blockers = list(session_blockers or [])
+    if session_blockers:
+        result["classification"] = CLASS_SESSION_OR_PROFILE_MISMATCH
+        result["mark_final_allowed"] = False
+        result["next_action"] = (
+            "reconnect matching prgs-reviewer session or wait for lock TTL; "
+            "do not use gitea_authorize_review_correction as unlock; "
+            "do not delete session-state files"
+        )
+        result["reasons"].extend(session_blockers)
+        return result
+
+    stored = (
+        (lock.get("profile_identity") or lock.get("session_profile_lock") or "")
+        .strip()
+    )
+    active = (active_profile_identity or "").strip()
+    if active and stored and active != stored:
+        result["classification"] = CLASS_SESSION_OR_PROFILE_MISMATCH
+        result["mark_final_allowed"] = False
+        result["next_action"] = (
+            f"switch to profile '{stored}' or wait for moot cleanup; "
+            "do not use correction authorization as a generic unlock"
+        )
+        result["reasons"].append(
+            f"lock profile '{stored}' does not match active '{active}'"
+        )
+        return result
+
+    last = last_terminal_mutation(lock)
+    if last is None:
+        ready_pr = lock.get("ready_pr_number")
+        ready_head = normalize_head_sha(lock.get("ready_expected_head_sha"))
+        if (
+            target_pr_number is not None
+            and ready_pr == target_pr_number
+            and ready_head
+            and target_head
+            and heads_equal(ready_head, target_head)
+            and lock.get("final_review_decision_ready")
+        ):
+            result["classification"] = CLASS_IN_PROGRESS_SAME_HEAD
+            result["mark_final_allowed"] = True  # idempotent re-mark
+            result["next_action"] = (
+                "gitea_submit_pr_review with final_review_decision_ready=true "
+                "for the ready PR/head"
+            )
+            result["reasons"].append(
+                "final decision already marked ready for this PR/head (idempotent)"
+            )
+            return result
+        result["classification"] = CLASS_READY_FOR_TARGET
+        result["next_action"] = "gitea_mark_final_review_decision then submit"
+        result["reasons"].append("no terminal live mutations; mark_final allowed")
+        return result
+
+    last_pr = last.get("pr_number")
+    last_action = last.get("action")
+    locked_head = mutation_head_sha(last, lock)
+    result["last_terminal_pr"] = last_pr
+    result["last_terminal_action"] = last_action
+    result["locked_head_sha"] = locked_head
+    result["evidence"] = {
+        "last_review_id": last.get("review_id"),
+        "live_mutations_count": len(lock.get("live_mutations") or []),
+    }
+
+    if correction_applies(
+        lock, pr_number=target_pr_number, expected_head_sha=target_head
+    ):
+        result["classification"] = CLASS_CORRECTION_ACTIVE
+        result["mark_final_allowed"] = True
+        result["next_action"] = (
+            "gitea_mark_final_review_decision for the correction-scoped PR/head, "
+            "then submit once"
+        )
+        result["reasons"].append(
+            f"scoped correction active for PR #{lock.get('correction_pr_number')}"
+        )
+        return result
+
+    # Live state of last terminal PR when provided.
+    live = None
+    if pr_lookup_error:
+        result["classification"] = CLASS_AMBIGUOUS
+        result["mark_final_allowed"] = False
+        result["next_action"] = (
+            "retry diagnosis after live PR state is available; "
+            "fail closed under ambiguous evidence"
+        )
+        result["reasons"].append(f"live PR lookup failed: {pr_lookup_error}")
+        return result
+    if pr_live is not None:
+        live = classify_pr_live_state(pr_live)
+        result["evidence"]["last_terminal_pr_state"] = live.get("pr_state")
+        result["evidence"]["last_terminal_pr_merged"] = live.get("pr_merged")
+        if live.get("pr_merged_or_closed"):
+            result["classification"] = CLASS_MOOT_MERGED_CLOSED
+            result["cleanup_allowed"] = True
+            result["mark_final_allowed"] = target_pr_number is not None and (
+                int(last_pr) != int(target_pr_number)
+                if last_pr is not None and target_pr_number is not None
+                else True
+            )
+            result["next_action"] = (
+                "gitea_cleanup_stale_review_decision_lock(apply=true, "
+                f"expected_terminal_pr={last_pr}) then mark_final on the target PR"
+            )
+            result["reasons"].append(
+                f"terminal on PR #{last_pr} is moot (merged/closed); cleanup allowed"
+            )
+            return result
+
+    if target_pr_number is None:
+        result["classification"] = CLASS_AMBIGUOUS
+        result["mark_final_allowed"] = False
+        result["next_action"] = "re-run diagnosis with target_pr_number"
+        result["reasons"].append("target_pr_number required for open-PR classification")
+        return result
+
+    if last_pr is not None and int(last_pr) != int(target_pr_number):
+        # Cross-PR isolation: foreign terminal is history, not a block.
+        result["classification"] = CLASS_FOREIGN_PR_TERMINAL
+        result["mark_final_allowed"] = True
+        result["correction_eligible"] = False  # must not use correction to "unlock"
+        result["next_action"] = (
+            f"gitea_mark_final_review_decision(pr_number={target_pr_number}, ...) "
+            f"— foreign terminal on PR #{last_pr} does not block (#693); "
+            "do not call gitea_authorize_review_correction for cross-PR unlock"
+        )
+        result["reasons"].append(
+            f"last terminal is {last_action} on foreign open/closed PR #{last_pr}; "
+            f"target PR #{target_pr_number} is isolated (#693)"
+        )
+        return result
+
+    # Same PR as target.
+    if (
+        locked_head
+        and target_head
+        and not heads_equal(locked_head, target_head)
+    ):
+        result["classification"] = CLASS_STALE_SUPERSEDED_HEAD
+        result["mark_final_allowed"] = True
+        result["next_action"] = (
+            f"gitea_mark_final_review_decision with expected_head_sha="
+            f"{target_head} (#620 same-PR new head)"
+        )
+        result["reasons"].append(
+            f"terminal {last_action} on head {locked_head[:12]}…; target head "
+            f"{target_head[:12]}… — fresh formal review allowed without cleanup"
+        )
+        return result
+
+    # Same PR + same head (or missing head comparison) — active terminal.
+    result["classification"] = CLASS_TERMINAL_ACTIVE_SAME_HEAD
+    result["mark_final_allowed"] = False
+    result["correction_eligible"] = True
+    result["next_action"] = (
+        "if the prior verdict was mistaken: gitea_authorize_review_correction "
+        f"with prior_review_id={last.get('review_id')}, "
+        f"prior_review_state={last_action}, target_pr_number={target_pr_number}, "
+        "operator_authorized=true, then re-mark once; "
+        "if the prior verdict is correct: stop and hand off (do not duplicate)"
+    )
+    result["reasons"].append(
+        f"terminal {last_action} already recorded for PR #{target_pr_number} "
+        f"at this head; same-head duplicate blocked (#332/#620)"
+    )
+    return result
+
+
+def build_precise_mark_final_failure(
+    classification: dict[str, Any],
+) -> dict[str, Any]:
+    """One precise recovery instruction + durable issue handoff payload (#693 AC4/8)."""
+    cls = classification.get("classification")
+    next_action = classification.get("next_action") or (
+        "stop and create a durable Gitea issue with lock evidence"
+    )
+    reasons = list(classification.get("reasons") or [])
+    reasons.append(f"exact_next_action: {next_action}")
+    handoff = {
+        "required": cls
+        in (
+            CLASS_AMBIGUOUS,
+            CLASS_SESSION_OR_PROFILE_MISMATCH,
+            CLASS_TERMINAL_ACTIVE_SAME_HEAD,
+        )
+        and not classification.get("mark_final_allowed"),
+        "title_hint": (
+            "Review-decision-lock recovery failed during formal review"
+        ),
+        "classification": cls,
+        "exact_next_action": next_action,
+        "owner_profile_identity": classification.get("owner_profile_identity"),
+        "last_terminal_pr": classification.get("last_terminal_pr"),
+        "last_terminal_action": classification.get("last_terminal_action"),
+        "locked_head_sha": classification.get("locked_head_sha"),
+        "target_pr_number": classification.get("target_pr_number"),
+        "target_head_sha": classification.get("target_head_sha"),
+        "evidence": classification.get("evidence") or {},
+        "instruction": (
+            "If recovery cannot complete with the exact_next_action above, "
+            "create or update a durable Gitea issue with this payload and stop; "
+            "do not delete session-state files; do not misuse correction as unlock."
+        ),
+    }
+    return {
+        "reasons": reasons,
+        "classification": cls,
+        "exact_next_action": next_action,
+        "durable_issue_handoff": handoff,
+    }
+
+
+def build_correction_audit_record(
+    *,
+    prior_review_id: int | None,
+    prior_review_state: str | None,
+    target_pr_number: int | None,
+    target_head_sha: str | None,
+    reason: str | None,
+    actor_username: str | None,
+    profile_name: str | None,
+    authorized: bool,
+) -> dict[str, Any]:
+    """Structured durable audit for review-correction authorization (#693)."""
+    return {
+        "event": "review_correction_authorization",
+        "issue_ref": "#693",
+        "authorized": bool(authorized),
+        "timestamp": datetime.now(timezone.utc).isoformat(),
+        "actor_username": actor_username,
+        "profile_name": profile_name,
+        "prior_review_id": prior_review_id,
+        "prior_review_state": prior_review_state,
+        "target_pr_number": target_pr_number,
+        "target_head_sha": normalize_head_sha(target_head_sha),
+        "reason": reason,
+        "scope": "same_pr_head_only",
+    }
+
+
+def format_correction_audit_comment(audit: dict[str, Any]) -> str:
+    """Markdown body for a thread-visible correction authorization audit."""
+    status = "AUTHORIZED" if audit.get("authorized") else "DENIED"
+    lines = [
+        "## Review correction authorization audit (#693)",
+        "",
+        f"Status: **{status}**",
+        "",
+        f"- actor: `{audit.get('actor_username')}`",
+        f"- profile: `{audit.get('profile_name')}`",
+        f"- timestamp: `{audit.get('timestamp')}`",
+        f"- prior_review_id: `{audit.get('prior_review_id')}`",
+        f"- prior_review_state: `{audit.get('prior_review_state')}`",
+        f"- target_pr: `#{audit.get('target_pr_number')}`",
+        f"- target_head_sha: `{audit.get('target_head_sha')}`",
+        f"- reason: {audit.get('reason')}",
+        f"- scope: `{audit.get('scope')}` (cannot unlock a different PR)",
+        "",
+        "This is **not** a generic decision-lock unlock. Cross-PR recovery "
+        "uses isolation (#693) or moot cleanup (#594), never correction.",
+    ]
+    return "\n".join(lines)
+
+
+def assess_open_pr_decision_lock_recovery(
+    lock: dict | None,
+    *,
+    target_pr_number: int,
+    target_head_sha: str | None,
+    last_terminal_pr_live: dict | None = None,
+    pr_lookup_error: str | None = None,
+    active_profile_identity: str | None = None,
+    session_blockers: list[str] | None = None,
+    controller_recovery_authorized: bool = False,
+) -> dict[str, Any]:
+    """Assess guarded recovery for decision locks affecting an open target PR (#693).
+
+    Recovery here means *workflow recovery routing*, not necessarily lock wipe.
+    Foreign-PR terminals are recoverable by isolation (mark_final allowed).
+    Moot terminals use #594 cleanup. Same-head active terminals refuse wipe.
+    """
+    classification = classify_review_decision_lock(
+        lock,
+        target_pr_number=target_pr_number,
+        target_head_sha=target_head_sha,
+        pr_live=last_terminal_pr_live,
+        pr_lookup_error=pr_lookup_error,
+        active_profile_identity=active_profile_identity,
+        session_blockers=session_blockers,
+    )
+    cls = classification["classification"]
+    recovery_allowed = False
+    recovery_mode = None
+    if cls == CLASS_FOREIGN_PR_TERMINAL:
+        recovery_allowed = True
+        recovery_mode = "cross_pr_isolation"
+    elif cls == CLASS_STALE_SUPERSEDED_HEAD:
+        recovery_allowed = True
+        recovery_mode = "same_pr_new_head"
+    elif cls == CLASS_MOOT_MERGED_CLOSED:
+        recovery_allowed = True
+        recovery_mode = "moot_cleanup"
+    elif cls == CLASS_READY_FOR_TARGET:
+        recovery_allowed = True
+        recovery_mode = "none_required"
+    elif cls == CLASS_CORRECTION_ACTIVE:
+        recovery_allowed = True
+        recovery_mode = "scoped_correction"
+    elif cls == CLASS_TERMINAL_ACTIVE_SAME_HEAD:
+        recovery_allowed = False
+        recovery_mode = "correction_only_if_mistake"
+    else:
+        recovery_allowed = False
+        recovery_mode = "fail_closed"
+
+    if recovery_mode == "moot_cleanup" and not controller_recovery_authorized:
+        # Cleanup still needs apply + capability; assess reports path.
+        pass
+
+    return {
+        **classification,
+        "recovery_allowed": recovery_allowed,
+        "recovery_mode": recovery_mode,
+        "controller_recovery_authorized": bool(controller_recovery_authorized),
+        "manual_file_delete_forbidden": True,
+        "correction_as_generic_unlock_forbidden": True,
+    }
+
+
 # ---------------------------------------------------------------------------
 # #709 cross-profile cleanup, overwrite protection, recovery provenance
 # ---------------------------------------------------------------------------

task_capability_map.py

@@ -60,18 +60,57 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
         "permission": "gitea.pr.close",
         "role": "author",
     },
+    # Non-closing PR metadata edits (title/body/base). Closing uses close_pr.
+    "edit_pr": {
+        "permission": "gitea.pr.create",
+        "role": "author",
+    },
+    "gitea_edit_pr": {
+        "permission": "gitea.pr.create",
+        "role": "author",
+    },
     "address_pr_change_requests": {
         "permission": "gitea.branch.push",
         "role": "author",
     },
+    # PR synchronization lifecycle: assess is read-only (any role with gitea.read);
+    # update-by-merge is author-only and mutates the PR head via Gitea API.
+    "assess_pr_sync_status": {
+        "permission": "gitea.read",
+        "role": "author",
+    },
+    "gitea_assess_pr_sync_status": {
+        "permission": "gitea.read",
+        "role": "author",
+    },
+    "update_pr_branch_by_merge": {
+        "permission": "gitea.branch.push",
+        "role": "author",
+    },
+    "gitea_update_pr_branch_by_merge": {
+        "permission": "gitea.branch.push",
+        "role": "author",
+    },
     "review_pr": {
         "permission": "gitea.pr.review",
         "role": "reviewer",
     },
+    "submit_pr_review": {
+        "permission": "gitea.pr.review",
+        "role": "reviewer",
+    },
     "merge_pr": {
         "permission": "gitea.pr.merge",
         "role": "merger",
     },
+    "acquire_reviewer_pr_lease": {
+        "permission": "gitea.pr.comment",
+        "role": "reviewer",
+    },
+    "gitea_acquire_reviewer_pr_lease": {
+        "permission": "gitea.pr.comment",
+        "role": "reviewer",
+    },
     # #695 AC8: controller quarantine of contaminated formal reviews.
     # Apply path posts an append-only forensic audit comment (pr.comment).
     "quarantine_contaminated_review": {
@@ -86,6 +125,18 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
         "permission": "gitea.pr.comment",
         "role": "merger",
     },
+    "gitea_adopt_merger_pr_lease": {
+        "permission": "gitea.pr.comment",
+        "role": "merger",
+    },
+    "acquire_merger_pr_lease": {
+        "permission": "gitea.pr.comment",
+        "role": "merger",
+    },
+    "gitea_acquire_merger_pr_lease": {
+        "permission": "gitea.pr.comment",
+        "role": "merger",
+    },
     # #691: guarded non-owner cleanup of obsolete comment-backed reviewer leases.
     # Apply path posts lease release + audit comments (gitea.pr.comment).
     "cleanup_obsolete_reviewer_comment_lease": {
@@ -127,6 +178,23 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
         "permission": "gitea.pr.review",
         "role": "reviewer",
     },
+    # #693: read-only classification of durable decision locks (incl. open PRs).
+    "diagnose_review_decision_lock": {
+        "permission": "gitea.read",
+        "role": "reviewer",
+    },
+    "gitea_diagnose_review_decision_lock": {
+        "permission": "gitea.read",
+        "role": "reviewer",
+    },
+    "authorize_review_correction": {
+        "permission": "gitea.pr.review",
+        "role": "reviewer",
+    },
+    "gitea_authorize_review_correction": {
+        "permission": "gitea.pr.review",
+        "role": "reviewer",
+    },
     # #709: truthful absence-of-proof recovery (server-side auth + record + consume).
     # Dedicated mutation capability — gitea.read is insufficient (review 434 F1).
     "issue_irrecoverable_provenance_authorization": {
@@ -330,32 +398,6 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
     },
 }
 
-# #723 AC1/AC5: tasks where permission alone is NOT enough — the profile's
-# role kind must also match. Shared by ``gitea_resolve_task_capability`` and
-# the runtime-context capability report so the two can never disagree about
-# which tasks are role-exclusive (incident #722: runtime context said
-# review_pr was allowed while the resolver fail-closed the same session).
-ROLE_EXCLUSIVE_TASKS: frozenset[str] = frozenset({
-    "review_pr",
-    "approve_pr",
-    "request_changes_pr",
-    "blind_pr_queue_review",
-    "pr_queue_cleanup",
-    "pr-queue-cleanup",
-    "merge_pr",
-    "create_branch",
-    "push_branch",
-    "create_pr",
-    "commit_files",
-    "gitea_commit_files",
-    "address_pr_change_requests",
-    "delete_branch",
-    "cleanup_merged_pr_branch",
-    "reconciliation_cleanup",
-    "work_issue",
-    "work-issue",
-})
-
 # Issue-mutating MCP tools and their resolver task keys.
 ISSUE_MUTATION_TOOL_TASKS: dict[str, str] = {
     "gitea_create_issue": "create_issue",

tests/test_head_scoped_review_decision_lock.py

@@ -24,6 +24,10 @@ HEAD_C = "c" * 40
 
 
 def _lock(mutations=None, correction=False, ready_pr=None, ready_head=None, ready_action=None):
+    mutations = list(mutations or [])
+    corr_pr = None
+    if correction and mutations:
+        corr_pr = mutations[-1].get("pr_number")
     return {
         "task": "review_pr",
         "remote": "prgs",
@@ -40,9 +44,11 @@ def _lock(mutations=None, correction=False, ready_pr=None, ready_head=None, read
         "ready_remote": "prgs" if ready_pr else None,
         "ready_org": "Scaled-Tech-Consulting" if ready_pr else None,
         "ready_repo": "Gitea-Tools" if ready_pr else None,
-        "live_mutations": list(mutations or []),
+        "live_mutations": mutations,
         "correction_authorized": correction,
-        "correction_reason": None,
+        "correction_reason": "test" if correction else None,
+        "correction_pr_number": corr_pr,
+        "correction_head_sha": None,
     }
 
 
@@ -174,12 +180,13 @@ class TestHardStopHeadScope(unittest.TestCase):
         self.assertTrue(reasons)
         self.assertIn("#332", reasons[0])
 
-    def test_rc_head_a_blocks_other_pr(self):
+    def test_rc_head_a_allows_other_pr_via_cross_pr_isolation(self):
+        """#693: foreign-PR terminal must not hard-stop mark_ready on another PR."""
         _seed([RC_619_HEAD_A], ready_pr=619, ready_head=HEAD_A)
         reasons = mcp_server.terminal_review_hard_stop_reasons(
             700, "mark_ready", expected_head_sha=HEAD_B
         )
-        self.assertTrue(reasons)
+        self.assertEqual(reasons, [])
 
     def test_legacy_619_ledger_allows_new_head(self):
         """PR #619-style durable lock without mutation head_sha."""

tests/test_issue_691_obsolete_reviewer_lease_cleanup.py

@@ -561,12 +561,29 @@ class TestIssue691SupersededHeadCleanup(unittest.TestCase):
         self.assertTrue(any("owns the lease" in r for r in result["reasons"]))
 
     def test_superseded_without_terminal_review_denied(self):
+        # Pre-expiry the missing terminal review is ambiguous (fail closed);
+        # post-expiry with unknown owner evidence it is a crash-orphan (#702)
+        # that still fails closed until owner-process exit is proven.
         now = datetime(2026, 7, 13, 5, 0, 0, tzinfo=timezone.utc)
         expires = datetime(2026, 7, 13, 4, 23, 0, tzinfo=timezone.utc)
         claim = _lease_comment(expires_at=expires, last_activity=expires - timedelta(hours=1))
-        result = _assess([claim], formal_reviews=[], now=now)
+        result = _assess(
+            [claim], formal_reviews=[], now=now, owner_process_alive=None
+        )
         self.assertFalse(result["cleanup_allowed"])
-        self.assertEqual(result["classification"], "ambiguous_conflicting_evidence")
+        self.assertEqual(
+            result["classification"], "orphaned_expired_superseded_head"
+        )
+
+        fresh_expires = now + timedelta(minutes=60)
+        fresh = _lease_comment(
+            expires_at=fresh_expires, last_activity=now - timedelta(minutes=5)
+        )
+        pre_expiry = _assess([fresh], formal_reviews=[], now=now)
+        self.assertFalse(pre_expiry["cleanup_allowed"])
+        self.assertEqual(
+            pre_expiry["classification"], "ambiguous_conflicting_evidence"
+        )
 
 
 class TestIssue691DiagnoseClassifications(unittest.TestCase):

tests/test_issue_693_review_decision_lock_recovery.py

@@ -0,0 +1,427 @@
+"""Review-decision-lock recovery / cross-PR isolation (#693).
+
+Reproduces the PR #688 → PR #692 formal-review sequence:
+
+  * durable lock holds REQUEST_CHANGES on open PR #688 @ head A (review_id 425)
+  * reviewer leases PR #692 @ head B (6d86db…)
+  * mark_final for #692 must succeed without correction unlock
+  * cleanup of open #688 terminal remains forbidden (#594)
+  * authorize_review_correction for #688 cannot unlock #692
+  * eventual #692 APPROVE is unique and head-bound (review_id 426)
+  * mark_final is idempotent for same PR/action/head
+  * diagnosis returns one exact next_action + durable handoff payload
+"""
+
+from __future__ import annotations
+
+import os
+import unittest
+from unittest.mock import patch
+
+import mcp_server
+import review_workflow_load
+import stale_review_decision_lock as srdl
+
+# PR #688 / #692 incident fixtures (sanitized reconstruction from issue #693)
+HEAD_688 = "c7a444eb4b41cf916fdbd20a4999ffd78af496d0"
+HEAD_692 = "6d86db793bbdfaed16bd54d93171f1dd66f234ca"
+PR_A = 688
+PR_B = 692
+REVIEW_ID_A = 425
+REVIEW_ID_B = 426
+
+RC_688 = {
+    "pr_number": PR_A,
+    "action": "request_changes",
+    "review_id": REVIEW_ID_A,
+    "review_state": "request_changes",
+    "head_sha": HEAD_688,
+}
+APPROVE_692 = {
+    "pr_number": PR_B,
+    "action": "approve",
+    "review_id": REVIEW_ID_B,
+    "review_state": "approve",
+    "head_sha": HEAD_692,
+}
+
+
+def _lock(mutations=None, **kwargs):
+    from datetime import datetime, timezone
+
+    now = datetime.now(timezone.utc).isoformat()
+    base = {
+        "task": "review_pr",
+        "remote": "prgs",
+        "org": "Scaled-Tech-Consulting",
+        "repo": "Gitea-Tools",
+        "session_pid": os.getpid(),
+        "session_profile": "prgs-reviewer",
+        "session_profile_lock": "prgs-reviewer",
+        "profile_identity": "prgs-reviewer",
+        "final_review_decision_ready": False,
+        "ready_pr_number": None,
+        "ready_action": None,
+        "ready_expected_head_sha": None,
+        "ready_remote": None,
+        "ready_org": None,
+        "ready_repo": None,
+        "live_mutations": list(mutations or []),
+        "correction_authorized": False,
+        "correction_reason": None,
+        "correction_pr_number": None,
+        "correction_head_sha": None,
+        "updated_at": now,
+        "recorded_at": now,
+        "writer_pid": os.getpid(),
+    }
+    base.update(kwargs)
+    # Ensure TTL fields stay fresh unless the caller is testing expiry.
+    if "recorded_at" not in kwargs:
+        base["recorded_at"] = now
+    if "updated_at" not in kwargs:
+        base["updated_at"] = now
+    return base
+
+
+REVIEWER_ENV = {
+    "GITEA_PROFILE_NAME": "prgs-reviewer",
+    "GITEA_ALLOWED_OPERATIONS": (
+        "gitea.read,gitea.pr.review,gitea.pr.approve,"
+        "gitea.pr.request_changes,gitea.pr.comment"
+    ),
+    "GITEA_SESSION_PROFILE_LOCK": "prgs-reviewer",
+}
+
+REVIEWER_PROFILE = {
+    "profile_name": "prgs-reviewer",
+    "allowed_operations": [
+        "gitea.read",
+        "gitea.pr.review",
+        "gitea.pr.approve",
+        "gitea.pr.request_changes",
+        "gitea.pr.comment",
+    ],
+    "forbidden_operations": ["gitea.pr.merge"],
+}
+
+
+def _seed(mutations=None, **kwargs):
+    review_workflow_load.record_review_workflow_load(mcp_server.PROJECT_ROOT)
+    mcp_server._save_review_decision_lock(_lock(mutations, **kwargs))
+    mcp_server.gitea_load_review_workflow()
+
+
+def _no_lease():
+    return {"block": False, "reasons": [], "mutation_allowed": True}
+
+
+def _open_pr(number, head):
+    return {
+        "number": number,
+        "state": "open",
+        "merged": False,
+        "merged_at": None,
+        "head": {"sha": head},
+    }
+
+
+class TestCrossPrIsolationPure(unittest.TestCase):
+    def test_foreign_terminal_does_not_block_mark_boundary(self):
+        lock = _lock([RC_688])
+        self.assertFalse(
+            srdl.prior_live_mutations_block_boundary(
+                lock, pr_number=PR_B, expected_head_sha=HEAD_692
+            )
+        )
+        self.assertTrue(
+            srdl.prior_live_mutations_block_boundary(
+                lock, pr_number=PR_A, expected_head_sha=HEAD_688
+            )
+        )
+
+    def test_terminal_boundary_allows_fresh_decision_on_other_pr(self):
+        lock = _lock([RC_688])
+        self.assertTrue(
+            srdl.terminal_boundary_allows_fresh_decision(
+                lock,
+                pr_number=PR_B,
+                expected_head_sha=HEAD_692,
+                operation="mark_ready",
+            )
+        )
+        self.assertFalse(
+            srdl.terminal_boundary_allows_fresh_decision(
+                lock,
+                pr_number=PR_A,
+                expected_head_sha=HEAD_688,
+                operation="mark_ready",
+            )
+        )
+
+    def test_unscoped_correction_does_not_apply(self):
+        lock = _lock([RC_688], correction_authorized=True, correction_reason="x")
+        # Missing correction_pr_number → fail closed (#693)
+        self.assertFalse(
+            srdl.correction_applies(lock, pr_number=PR_A, expected_head_sha=HEAD_688)
+        )
+
+    def test_scoped_correction_only_for_named_pr(self):
+        lock = _lock(
+            [RC_688],
+            correction_authorized=True,
+            correction_reason="mistake",
+            correction_pr_number=PR_A,
+            correction_head_sha=HEAD_688,
+        )
+        self.assertTrue(
+            srdl.correction_applies(lock, pr_number=PR_A, expected_head_sha=HEAD_688)
+        )
+        self.assertFalse(
+            srdl.correction_applies(lock, pr_number=PR_B, expected_head_sha=HEAD_692)
+        )
+
+
+class TestClassification(unittest.TestCase):
+    def test_foreign_pr_terminal_classification(self):
+        lock = _lock([RC_688])
+        c = srdl.classify_review_decision_lock(
+            lock,
+            target_pr_number=PR_B,
+            target_head_sha=HEAD_692,
+            pr_live=_open_pr(PR_A, HEAD_688),
+            active_profile_identity="prgs-reviewer",
+        )
+        self.assertEqual(c["classification"], srdl.CLASS_FOREIGN_PR_TERMINAL)
+        self.assertTrue(c["mark_final_allowed"])
+        self.assertFalse(c["correction_eligible"])
+        self.assertIn("do not call gitea_authorize_review_correction", c["next_action"])
+
+    def test_same_head_terminal_classification(self):
+        lock = _lock([RC_688])
+        c = srdl.classify_review_decision_lock(
+            lock,
+            target_pr_number=PR_A,
+            target_head_sha=HEAD_688,
+            pr_live=_open_pr(PR_A, HEAD_688),
+        )
+        self.assertEqual(c["classification"], srdl.CLASS_TERMINAL_ACTIVE_SAME_HEAD)
+        self.assertFalse(c["mark_final_allowed"])
+        self.assertTrue(c["correction_eligible"])
+
+    def test_precise_failure_includes_durable_handoff(self):
+        lock = _lock([RC_688])
+        c = srdl.classify_review_decision_lock(
+            lock, target_pr_number=PR_A, target_head_sha=HEAD_688
+        )
+        fail = srdl.build_precise_mark_final_failure(c)
+        self.assertIn("exact_next_action", fail)
+        self.assertIn("durable_issue_handoff", fail)
+        self.assertTrue(fail["durable_issue_handoff"]["required"])
+
+
+class TestPr692Sequence(unittest.TestCase):
+    def setUp(self):
+        self._env = patch.dict(os.environ, REVIEWER_ENV, clear=False)
+        self._env.start()
+        self._prof = patch.object(
+            mcp_server, "get_profile", return_value=REVIEWER_PROFILE
+        )
+        self._prof.start()
+
+    def tearDown(self):
+        self._prof.stop()
+        self._env.stop()
+        mcp_server._save_review_decision_lock(None)
+        review_workflow_load.clear_review_workflow_load()
+
+    def test_mark_final_692_succeeds_with_open_688_terminal(self):
+        _seed([RC_688], writer_pid=25883, session_pid=60615)
+        with patch("mcp_server._list_pr_lease_comments", return_value=[]), patch(
+            "mcp_server._pr_work_lease_reviewer_block", return_value=_no_lease()
+        ), patch.object(
+            mcp_server,
+            "gitea_check_pr_eligibility",
+            return_value={"head_sha": HEAD_692, "eligible": True},
+        ):
+            result = mcp_server.gitea_mark_final_review_decision(
+                pr_number=PR_B,
+                action="approve",
+                expected_head_sha=HEAD_692,
+                remote="prgs",
+                org="Scaled-Tech-Consulting",
+                repo="Gitea-Tools",
+            )
+        self.assertTrue(result["marked_ready"], result.get("reasons"))
+        lock = mcp_server._load_review_decision_lock()
+        self.assertEqual(lock["ready_pr_number"], PR_B)
+        self.assertEqual(
+            srdl.normalize_head_sha(lock["ready_expected_head_sha"]), HEAD_692
+        )
+        # Foreign terminal history preserved (non-destructive isolation).
+        self.assertEqual(lock["live_mutations"][0]["pr_number"], PR_A)
+
+    def test_mark_final_idempotent_same_binding(self):
+        _seed(
+            [RC_688],
+            final_review_decision_ready=True,
+            ready_pr_number=PR_B,
+            ready_action="approve",
+            ready_expected_head_sha=HEAD_692,
+            ready_remote="prgs",
+            ready_org="Scaled-Tech-Consulting",
+            ready_repo="Gitea-Tools",
+        )
+        with patch("mcp_server._list_pr_lease_comments", return_value=[]), patch(
+            "mcp_server._pr_work_lease_reviewer_block", return_value=_no_lease()
+        ), patch.object(
+            mcp_server,
+            "gitea_check_pr_eligibility",
+            return_value={"head_sha": HEAD_692, "eligible": True},
+        ):
+            result = mcp_server.gitea_mark_final_review_decision(
+                pr_number=PR_B,
+                action="approve",
+                expected_head_sha=HEAD_692,
+                remote="prgs",
+                org="Scaled-Tech-Consulting",
+                repo="Gitea-Tools",
+            )
+        self.assertTrue(result["marked_ready"])
+        self.assertTrue(result.get("idempotent"))
+
+    def test_cleanup_still_forbidden_while_688_open(self):
+        _seed([RC_688])
+        assessment = srdl.assess_stale_review_decision_lock(
+            mcp_server._load_review_decision_lock(),
+            pr_live=_open_pr(PR_A, HEAD_688),
+            active_profile_identity="prgs-reviewer",
+        )
+        self.assertFalse(assessment["cleanup_allowed"])
+        self.assertFalse(assessment["is_moot"])
+
+    def test_correction_cannot_unlock_different_pr(self):
+        _seed([RC_688])
+        r = mcp_server.gitea_authorize_review_correction(
+            prior_review_id=REVIEW_ID_A,
+            prior_review_state="request_changes",
+            reason="try unlock 692",
+            operator_authorized=True,
+            target_pr_number=PR_B,
+            remote="prgs",
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+            post_audit_comment=False,
+        )
+        self.assertFalse(r["authorized"])
+        self.assertTrue(any("different PR" in x for x in r["reasons"]))
+
+    def test_scoped_correction_for_same_pr_posts_audit(self):
+        _seed([RC_688])
+        with patch(
+            "mcp_server._authenticated_username", return_value="sysadmin"
+        ), patch(
+            "mcp_server.get_auth_header", return_value="Basic dGVzdDp0ZXN0"
+        ), patch(
+            "mcp_server._auth", return_value="Basic dGVzdDp0ZXN0"
+        ), patch("mcp_server.api_request", return_value={"id": 99901}):
+            r = mcp_server.gitea_authorize_review_correction(
+                prior_review_id=REVIEW_ID_A,
+                prior_review_state="request_changes",
+                reason="mistaken RC wording",
+                operator_authorized=True,
+                target_pr_number=PR_A,
+                expected_head_sha=HEAD_688,
+                remote="prgs",
+                org="Scaled-Tech-Consulting",
+                repo="Gitea-Tools",
+                post_audit_comment=True,
+            )
+        self.assertTrue(r["authorized"], r.get("reasons"))
+        self.assertEqual(r.get("correction_pr_number"), PR_A)
+        self.assertEqual(
+            r.get("audit_comment_id"),
+            99901,
+            r.get("audit_comment_skipped") or r.get("audit_comment_error") or r,
+        )
+
+    def test_diagnose_foreign_terminal_next_action(self):
+        _seed([RC_688])
+        with patch(
+            "mcp_server._authenticated_username", return_value="sysadmin"
+        ), patch(
+            "mcp_server.get_auth_header", return_value="Basic dGVzdDp0ZXN0"
+        ), patch(
+            "mcp_server._auth", return_value="Basic dGVzdDp0ZXN0"
+        ), patch("mcp_server.api_request", return_value=_open_pr(PR_A, HEAD_688)):
+            d = mcp_server.gitea_diagnose_review_decision_lock(
+                target_pr_number=PR_B,
+                expected_head_sha=HEAD_692,
+                remote="prgs",
+                org="Scaled-Tech-Consulting",
+                repo="Gitea-Tools",
+            )
+        self.assertTrue(d["success"], d.get("reasons"))
+        self.assertEqual(d["classification"], srdl.CLASS_FOREIGN_PR_TERMINAL)
+        self.assertTrue(d["mark_final_allowed"])
+        self.assertIn("mark_final", d["exact_next_action"])
+        self.assertTrue(d["correction_as_generic_unlock_forbidden"])
+
+    def test_approval_ledger_unique_and_bound_after_sequence(self):
+        """After mark_final + record mutation, ledger binds unique approve to 692 head."""
+        _seed([RC_688])
+        with patch("mcp_server._list_pr_lease_comments", return_value=[]), patch(
+            "mcp_server._pr_work_lease_reviewer_block", return_value=_no_lease()
+        ), patch.object(
+            mcp_server,
+            "gitea_check_pr_eligibility",
+            return_value={"head_sha": HEAD_692, "eligible": True},
+        ):
+            marked = mcp_server.gitea_mark_final_review_decision(
+                pr_number=PR_B,
+                action="approve",
+                expected_head_sha=HEAD_692,
+                remote="prgs",
+                org="Scaled-Tech-Consulting",
+                repo="Gitea-Tools",
+            )
+        self.assertTrue(marked["marked_ready"])
+        mcp_server.record_live_review_mutation(PR_B, "approve", REVIEW_ID_B)
+        lock = mcp_server._load_review_decision_lock()
+        terminals = [
+            m
+            for m in lock["live_mutations"]
+            if m.get("action") in srdl.TERMINAL_REVIEW_ACTIONS
+            and m.get("pr_number") == PR_B
+        ]
+        self.assertEqual(len(terminals), 1)
+        self.assertEqual(terminals[0]["review_id"], REVIEW_ID_B)
+        self.assertEqual(
+            srdl.normalize_head_sha(terminals[0].get("head_sha")), HEAD_692
+        )
+        # Same-head duplicate still blocked.
+        self.assertTrue(
+            srdl.prior_live_mutations_block_boundary(
+                lock, pr_number=PR_B, expected_head_sha=HEAD_692
+            )
+        )
+
+
+class TestSessionRestartWriterPid(unittest.TestCase):
+    def tearDown(self):
+        mcp_server._save_review_decision_lock(None)
+        review_workflow_load.clear_review_workflow_load()
+
+    def test_writer_pid_change_does_not_inherit_foreign_block(self):
+        # session_pid 60615 origin, writer later 25883 — still foreign isolation.
+        _seed([RC_688], session_pid=60615, writer_pid=25883)
+        self.assertEqual(
+            mcp_server.terminal_review_hard_stop_reasons(
+                PR_B, "mark_ready", expected_head_sha=HEAD_692
+            ),
+            [],
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_issue_702_review_findings_f1_f6.py

@@ -0,0 +1,693 @@
+"""Regression tests for the PR #703 REQUEST_CHANGES findings F1–F6 (#702).
+
+Formal review at head 889931d independently reproduced six defects:
+
+F1 — stale-binding recovery ran only AFTER the terminal launcher probe in
+     ``gitea_resolve_task_capability``; a worktree removed mid-session wedged
+     every mutation-task resolution on 'missing cwd' before recovery.
+F2 — ``_resolve_preflight_workspace_path`` skipped the existence checks the
+     canonical mutation context applies; a dead binding could produce empty
+     porcelain and read as a clean workspace.
+F3 — ``orphaned_expired_superseded_head`` cleanup accepted unknown
+     ``worktree_exists``/``worktree_clean`` evidence.
+F4 — the 4-hour session-state TTL silently discarded recovery-critical
+     crash-orphan shadow evidence.
+F5 — the single same-profile shadow slot let concurrent sessions overwrite
+     each other's crash evidence.
+F6 — the environment was cleared BEFORE the audit record was persisted, and
+     audit-write failure was swallowed.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import shutil
+import sys
+import tempfile
+import unittest
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+from unittest.mock import patch
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+import mcp_session_state as mss  # noqa: E402
+import namespace_workspace_binding as nwb  # noqa: E402
+import reviewer_pr_lease as leases  # noqa: E402
+import stale_binding_recovery as sbr  # noqa: E402
+import mcp_server  # noqa: E402
+
+REPO = "Scaled-Tech-Consulting/Gitea-Tools"
+OLD_HEAD = "b" * 40
+NEW_HEAD = "6" * 40
+
+CONFIG = {
+    "version": 2,
+    "contexts": {
+        "ctx": {
+            "enabled": True,
+            "gitea": {"enabled": True, "base_url": "https://gitea.example.com"},
+        }
+    },
+    "profiles": {
+        "prgs-author": {
+            "enabled": True,
+            "context": "ctx",
+            "role": "author",
+            "username": "jcwalker3",
+            "auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"},
+            "allowed_operations": [
+                "gitea.read", "gitea.issue.create", "gitea.issue.comment",
+                "gitea.pr.create", "gitea.branch.push",
+            ],
+            "forbidden_operations": [
+                "gitea.pr.approve", "gitea.pr.merge", "gitea.pr.review",
+            ],
+            "execution_profile": "prgs-author",
+        },
+    },
+    "rules": {"allow_runtime_switching": False},
+}
+
+
+def _now() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+class _CapabilityHarness(unittest.TestCase):
+    """Shared config/env plumbing for resolve-capability integration tests."""
+
+    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()
+        mcp_server._IDENTITY_CACHE.clear()
+        self._dir = tempfile.TemporaryDirectory()
+        self.config_path = os.path.join(self._dir.name, "profiles.json")
+        with open(self.config_path, "w", encoding="utf-8") as fh:
+            fh.write(json.dumps(CONFIG))
+
+    def tearDown(self):
+        self._remotes.stop()
+        mcp_server._IDENTITY_CACHE.clear()
+        self._dir.cleanup()
+
+    def _env(self, **extra):
+        env = {
+            "GITEA_MCP_CONFIG": self.config_path,
+            "GITEA_MCP_PROFILE": "prgs-author",
+            "GITEA_TOKEN_AUTHOR": "author-pass",
+        }
+        env.update(extra)
+        return env
+
+    def _deleted_worktree(self) -> str:
+        path = tempfile.mkdtemp(prefix="gitea-removed-worktree-")
+        shutil.rmtree(path)
+        return path
+
+
+class TestF1RecoveryBeforeTerminalProbe(_CapabilityHarness):
+    """F1: recovery must run before the terminal cwd probe can wedge."""
+
+    @patch("mcp_server.api_request", return_value={"login": "jcwalker3"})
+    @patch("mcp_server.get_auth_header", return_value="token author-pass")
+    def test_removed_worktree_recovers_before_probe(self, _auth, _api):
+        missing = self._deleted_worktree()
+        env = self._env(
+            GITEA_ACTIVE_WORKTREE=missing,
+            GITEA_FORCE_TERMINAL_PROBE="1",
+            GITEA_FORCE_STALE_BINDING_RECOVERY="1",
+        )
+        with patch.dict(os.environ, env):
+            res = mcp_server.gitea_resolve_task_capability(
+                task="create_pr", remote="prgs"
+            )
+            # The provably-stale binding was cleared before the probe ran.
+            self.assertNotIn("GITEA_ACTIVE_WORKTREE", os.environ)
+        self.assertFalse(res.get("terminal_launcher_unhealthy"))
+        report = res.get("stale_binding_recovery") or {}
+        self.assertEqual(
+            report.get("classification"),
+            sbr.CLASSIFICATION_PROVABLY_STALE_MISSING_PATH,
+        )
+        self.assertTrue(report.get("recovery_performed"))
+        self.assertTrue(res.get("allowed_in_current_session"))
+
+    @patch("mcp_server.api_request", return_value={"login": "jcwalker3"})
+    @patch("mcp_server.get_auth_header", return_value="token author-pass")
+    def test_probe_block_still_carries_recovery_report(self, _auth, _api):
+        # Recovery disabled (test mode without the force flag) preserves the
+        # fail-closed probe block, but the block must now carry the
+        # classification evidence instead of hiding it.
+        missing = self._deleted_worktree()
+        env = self._env(
+            GITEA_ACTIVE_WORKTREE=missing,
+            GITEA_FORCE_TERMINAL_PROBE="1",
+        )
+        with patch.dict(os.environ, env):
+            res = mcp_server.gitea_resolve_task_capability(
+                task="create_pr", remote="prgs"
+            )
+            self.assertIn("GITEA_ACTIVE_WORKTREE", os.environ)
+        self.assertTrue(res.get("terminal_launcher_unhealthy"))
+        report = res.get("stale_binding_recovery") or {}
+        self.assertEqual(
+            report.get("classification"),
+            sbr.CLASSIFICATION_PROVABLY_STALE_MISSING_PATH,
+        )
+        self.assertFalse(report.get("recovery_performed"))
+
+    @patch("mcp_server.api_request", return_value={"login": "jcwalker3"})
+    @patch("mcp_server.get_auth_header", return_value="token author-pass")
+    def test_unverified_binding_is_never_cleared_by_reordering(self, _auth, _api):
+        # F1 must not weaken authorization: an existing, uncorroborated
+        # binding survives resolution untouched even with recovery forced.
+        existing = tempfile.mkdtemp(prefix="gitea-existing-worktree-")
+        self.addCleanup(shutil.rmtree, existing, ignore_errors=True)
+        env = self._env(
+            GITEA_ACTIVE_WORKTREE=existing,
+            GITEA_FORCE_TERMINAL_PROBE="1",
+            GITEA_FORCE_STALE_BINDING_RECOVERY="1",
+        )
+        with patch.dict(os.environ, env):
+            res = mcp_server.gitea_resolve_task_capability(
+                task="create_pr", remote="prgs"
+            )
+            self.assertEqual(os.environ.get("GITEA_ACTIVE_WORKTREE"), existing)
+        report = res.get("stale_binding_recovery") or {}
+        self.assertFalse(report.get("recovery_performed"))
+
+
+class TestF2PreflightPathVerification(_CapabilityHarness):
+    """F2: preflight and mutation contexts verify resolved paths alike."""
+
+    def test_preflight_and_mutation_context_agree_on_dead_binding(self):
+        missing = self._deleted_worktree()
+        env = self._env(GITEA_ACTIVE_WORKTREE=missing)
+        with patch.dict(os.environ, env):
+            preflight = mcp_server._resolve_preflight_workspace_path(None)
+            mutation = mcp_server._resolve_namespace_mutation_context(None)
+        self.assertEqual(preflight, mutation["workspace_path"])
+        self.assertNotEqual(preflight, os.path.realpath(missing))
+
+    def test_missing_explicit_worktree_never_reads_clean(self):
+        missing = self._deleted_worktree()
+        with patch.dict(os.environ, self._env()):
+            porcelain = mcp_server._get_workspace_porcelain(worktree_path=missing)
+        self.assertEqual(
+            porcelain, mcp_server.PREFLIGHT_WORKSPACE_UNAVAILABLE_PORCELAIN
+        )
+        # The sentinel parses as a tracked dirty entry — never clean.
+        self.assertTrue(mcp_server._parse_porcelain_entries(porcelain))
+
+    def test_workspace_deleted_during_evaluation_never_reads_clean(self):
+        # TOCTOU: resolution succeeded, then the path vanished before git ran.
+        missing = self._deleted_worktree()
+        with patch.dict(os.environ, self._env()):
+            with patch.object(
+                mcp_server,
+                "_resolve_preflight_workspace_path",
+                return_value=missing,
+            ):
+                porcelain = mcp_server._get_workspace_porcelain()
+        self.assertEqual(
+            porcelain, mcp_server.PREFLIGHT_WORKSPACE_UNAVAILABLE_PORCELAIN
+        )
+
+    def test_git_failure_never_reads_clean(self):
+        class _Failed:
+            returncode = 128
+            stdout = ""
+            stderr = "fatal: not a git repository"
+
+        with patch.dict(os.environ, self._env()):
+            with patch.object(
+                mcp_server.subprocess, "run", return_value=_Failed()
+            ):
+                porcelain = mcp_server._get_workspace_porcelain()
+        self.assertEqual(
+            porcelain, mcp_server.PREFLIGHT_WORKSPACE_UNAVAILABLE_PORCELAIN
+        )
+
+    def test_symlinked_binding_to_deleted_target_is_demoted(self):
+        target = tempfile.mkdtemp(prefix="gitea-symlink-target-")
+        holder = tempfile.mkdtemp(prefix="gitea-symlink-holder-")
+        self.addCleanup(shutil.rmtree, holder, ignore_errors=True)
+        link = os.path.join(holder, "worktree-link")
+        os.symlink(target, link)
+        shutil.rmtree(target)
+        demotions: list[str] = []
+        _path, source = nwb.resolve_namespace_workspace(
+            role_kind="reviewer",
+            process_project_root=os.getcwd(),
+            env={"GITEA_ACTIVE_WORKTREE": link},
+            demotions=demotions,
+            verify_paths=True,
+        )
+        self.assertEqual(source, "MCP server process root (default)")
+        self.assertEqual(len(demotions), 1)
+
+    def test_dead_binding_falls_through_to_live_role_binding(self):
+        # Path changes during evaluation: the dead candidate demotes at
+        # verification time and never resurrects over the live fallback.
+        alive = tempfile.mkdtemp(prefix="gitea-alive-worktree-")
+        self.addCleanup(shutil.rmtree, alive, ignore_errors=True)
+        missing = self._deleted_worktree()
+        path, source = nwb.resolve_namespace_workspace(
+            role_kind="reviewer",
+            process_project_root=os.getcwd(),
+            env={
+                "GITEA_ACTIVE_WORKTREE": missing,
+                "GITEA_REVIEWER_WORKTREE": alive,
+            },
+            verify_paths=True,
+        )
+        self.assertEqual(path, os.path.realpath(alive))
+        self.assertEqual(
+            source, "GITEA_REVIEWER_WORKTREE environment variable"
+        )
+
+
+class TestF3AffirmativeWorktreeEvidence(unittest.TestCase):
+    """F3: unknown worktree evidence must fail closed for crash orphans."""
+
+    def _comments(self) -> list[dict]:
+        stale = _now() - timedelta(hours=3)
+        body = leases.format_lease_body(
+            repo=REPO,
+            pr_number=701,
+            issue_number=699,
+            reviewer_identity="sysadmin",
+            profile="prgs-reviewer",
+            session_id="65664-4f248d213d60",
+            worktree="branches/review-pr-654",
+            phase="validation-start",
+            candidate_head=OLD_HEAD,
+            target_branch="master",
+            target_branch_sha="2" * 40,
+            last_activity=stale,
+            expires_at=stale + timedelta(minutes=120),
+        )
+        return [{
+            "id": 11131,
+            "body": body,
+            "user": {"login": "sysadmin"},
+            "author": "sysadmin",
+            "created_at": stale.isoformat(),
+            "updated_at": stale.isoformat(),
+        }]
+
+    def _assess(self, **kwargs):
+        defaults = dict(
+            pr_number=701,
+            current_head_sha=NEW_HEAD,
+            formal_reviews=[],
+            repo=REPO,
+            expected_repo=REPO,
+            requesting_session_id="fresh-controller",
+            controller_recovery_authorized=True,
+            owner_process_alive=False,
+        )
+        defaults.update(kwargs)
+        return leases.assess_obsolete_reviewer_comment_lease_cleanup(
+            self._comments(), **defaults
+        )
+
+    def test_unknown_worktree_evidence_fails_closed(self):
+        result = self._assess(worktree_exists=None, worktree_clean=None)
+        self.assertEqual(
+            result["classification"], "orphaned_expired_superseded_head"
+        )
+        self.assertFalse(result["cleanup_allowed"])
+        self.assertIn(
+            "affirmative safe evidence",
+            " ".join(result["fail_closed_reasons"]),
+        )
+
+    def test_unknown_cleanliness_with_existing_worktree_fails_closed(self):
+        result = self._assess(worktree_exists=True, worktree_clean=None)
+        self.assertFalse(result["cleanup_allowed"])
+
+    def test_affirmative_clean_worktree_is_eligible(self):
+        result = self._assess(worktree_exists=True, worktree_clean=True)
+        self.assertEqual(
+            result["classification"], "orphaned_expired_superseded_head"
+        )
+        self.assertTrue(result["cleanup_allowed"])
+
+    def test_affirmative_absent_worktree_is_eligible(self):
+        result = self._assess(worktree_exists=False, worktree_clean=None)
+        self.assertTrue(result["cleanup_allowed"])
+
+    def test_matrix_matches_diagnosis_gate(self):
+        # The cleanup matrix and the diagnosis path must agree: unknown
+        # evidence yields acquire-only, affirmative evidence yields cleanup.
+        diagnosis_unknown = leases.diagnose_reviewer_pr_lease_handoff(
+            self._comments(),
+            pr_number=701,
+            current_session_id=None,
+            current_reviewer_identity="fresh-reviewer",
+            current_head_sha=NEW_HEAD,
+            formal_reviews=[],
+            owner_process_alive=False,
+            worktree_exists=None,
+            worktree_clean=None,
+        )
+        assess_unknown = self._assess(worktree_exists=None, worktree_clean=None)
+        self.assertNotEqual(
+            diagnosis_unknown["next_action"],
+            "cleanup_obsolete_reviewer_comment_lease",
+        )
+        self.assertFalse(assess_unknown["cleanup_allowed"])
+
+
+class TestF4CrashShadowDurability(unittest.TestCase):
+    """F4: crash-orphan evidence survives the former 4h TTL boundary."""
+
+    SID = "82984-abcabcabcabc"
+
+    def setUp(self):
+        self._dir = tempfile.TemporaryDirectory()
+        self._env = patch.dict(
+            os.environ, {"GITEA_MCP_SESSION_STATE_DIR": self._dir.name}
+        )
+        self._env.start()
+        leases._SESSION_LEASE = None
+
+    def tearDown(self):
+        self._env.stop()
+        self._dir.cleanup()
+        leases._SESSION_LEASE = None
+
+    def _age_record(self, kind: str, hours: float, instance_id: str | None = None):
+        path = mss.state_file_path(kind=kind, instance_id=instance_id)
+        with open(path, encoding="utf-8") as fh:
+            envelope = json.load(fh)
+        stamp = (
+            (_now() - timedelta(hours=hours)).isoformat().replace("+00:00", "Z")
+        )
+        envelope["recorded_at"] = stamp
+        envelope["updated_at"] = stamp
+        envelope["payload"]["recorded_at"] = stamp
+        envelope["payload"]["updated_at"] = stamp
+        with open(path, "w", encoding="utf-8") as fh:
+            json.dump(envelope, fh)
+
+    def _save_shadow(self) -> None:
+        mss.save_state(
+            kind=mss.KIND_REVIEWER_SESSION_LEASE,
+            instance_id=self.SID,
+            payload={"session_id": self.SID, "owner_pid": os.getpid()},
+        )
+
+    def _load_shadow(self):
+        return mss.load_state(
+            kind=mss.KIND_REVIEWER_SESSION_LEASE, instance_id=self.SID
+        )
+
+    def test_shadow_loads_before_former_ttl_boundary(self):
+        self._save_shadow()
+        self._age_record(
+            mss.KIND_REVIEWER_SESSION_LEASE, hours=3.9, instance_id=self.SID
+        )
+        self.assertIsNotNone(self._load_shadow())
+
+    def test_shadow_loads_at_former_ttl_boundary(self):
+        self._save_shadow()
+        self._age_record(
+            mss.KIND_REVIEWER_SESSION_LEASE, hours=4.0, instance_id=self.SID
+        )
+        self.assertIsNotNone(self._load_shadow())
+
+    def test_shadow_loads_long_after_former_ttl_boundary(self):
+        self._save_shadow()
+        self._age_record(
+            mss.KIND_REVIEWER_SESSION_LEASE, hours=27.0, instance_id=self.SID
+        )
+        record = self._load_shadow()
+        self.assertIsNotNone(record)
+        self.assertEqual(record.get("session_id"), self.SID)
+
+    def test_stale_binding_audit_record_is_also_durable(self):
+        mss.save_state(
+            kind=mss.KIND_STALE_BINDING_RECOVERY,
+            payload={"cleared_env": "GITEA_ACTIVE_WORKTREE"},
+        )
+        self._age_record(mss.KIND_STALE_BINDING_RECOVERY, hours=27.0)
+        self.assertIsNotNone(
+            mss.load_state(kind=mss.KIND_STALE_BINDING_RECOVERY)
+        )
+
+    def test_non_recovery_kinds_keep_ttl(self):
+        mss.save_state(
+            kind=mss.KIND_WORKFLOW_LOAD, payload={"workflow_hash": "abc"}
+        )
+        self._age_record(mss.KIND_WORKFLOW_LOAD, hours=5.0)
+        self.assertIsNone(mss.load_state(kind=mss.KIND_WORKFLOW_LOAD))
+
+    def test_sanctioned_clear_is_the_terminal_reconciliation(self):
+        self._save_shadow()
+        self._age_record(
+            mss.KIND_REVIEWER_SESSION_LEASE, hours=27.0, instance_id=self.SID
+        )
+        mss.clear_state(
+            kind=mss.KIND_REVIEWER_SESSION_LEASE, instance_id=self.SID
+        )
+        self.assertIsNone(self._load_shadow())
+        self.assertEqual(
+            mss.list_states(kind=mss.KIND_REVIEWER_SESSION_LEASE), []
+        )
+
+    def test_crash_orphan_recovery_works_after_former_ttl(self):
+        # End to end: a crashed session's shadow still proves owner exit a
+        # day later.
+        leases.record_session_lease(
+            {
+                "pr_number": 701,
+                "session_id": self.SID,
+                "worktree": "branches/review-pr-654",
+                "candidate_head": OLD_HEAD,
+                "repo": REPO,
+                "comment_id": 11131,
+                "profile": "prgs-reviewer",
+                "reviewer_identity": "sysadmin",
+                "phase": "claimed",
+            },
+            lease_provenance={"source": "acquire", "comment_id": 11131},
+        )
+        leases._SESSION_LEASE = None  # simulated crash: no sanctioned clear
+        self._age_record(
+            mss.KIND_REVIEWER_SESSION_LEASE, hours=27.0, instance_id=self.SID
+        )
+        shadow = leases.load_session_lease_shadow(session_id=self.SID)
+        self.assertIsNotNone(shadow)
+        verdict = leases.assess_crashed_session_lease_orphan(
+            shadow,
+            lease={"session_id": self.SID},
+            current_pid=os.getpid() + 1,
+            pid_alive=False,
+        )
+        self.assertTrue(verdict["orphan_evidence"])
+        self.assertIs(verdict["owner_process_alive"], False)
+
+
+class TestF5ConcurrentSessionShadows(unittest.TestCase):
+    """F5: concurrent same-profile sessions keep distinct shadow records."""
+
+    SID_A = "11111-aaaaaaaaaaaa"
+    SID_B = "22222-bbbbbbbbbbbb"
+
+    def setUp(self):
+        self._dir = tempfile.TemporaryDirectory()
+        self._env = patch.dict(
+            os.environ, {"GITEA_MCP_SESSION_STATE_DIR": self._dir.name}
+        )
+        self._env.start()
+        leases._SESSION_LEASE = None
+
+    def tearDown(self):
+        self._env.stop()
+        self._dir.cleanup()
+        leases._SESSION_LEASE = None
+
+    def _lease(self, session_id: str, pr_number: int, head: str) -> dict:
+        return {
+            "pr_number": pr_number,
+            "session_id": session_id,
+            "worktree": f"branches/review-pr-{pr_number}-{session_id[:5]}",
+            "candidate_head": head,
+            "repo": REPO,
+            "comment_id": 11221,
+            "profile": "prgs-reviewer",
+            "reviewer_identity": "sysadmin",
+            "phase": "claimed",
+        }
+
+    def _record(self, lease: dict) -> None:
+        leases.record_session_lease(
+            lease,
+            lease_provenance={"source": "acquire", "comment_id": 11221},
+        )
+
+    def test_interleaved_sessions_do_not_overwrite_each_other(self):
+        self._record(self._lease(self.SID_A, 703, OLD_HEAD))
+        self._record(self._lease(self.SID_B, 701, NEW_HEAD))
+        # Session A heartbeats again after B wrote.
+        updated_a = self._lease(self.SID_A, 703, OLD_HEAD)
+        updated_a["phase"] = "validation-start"
+        self._record(updated_a)
+
+        shadow_a = leases.load_session_lease_shadow(session_id=self.SID_A)
+        shadow_b = leases.load_session_lease_shadow(session_id=self.SID_B)
+        self.assertEqual(shadow_a.get("session_id"), self.SID_A)
+        self.assertEqual(shadow_a.get("pr_number"), 703)
+        self.assertEqual(shadow_a.get("phase"), "validation-start")
+        self.assertEqual(shadow_b.get("session_id"), self.SID_B)
+        self.assertEqual(shadow_b.get("pr_number"), 701)
+        self.assertEqual(shadow_b.get("candidate_head"), NEW_HEAD)
+
+    def test_shadow_lookup_never_misattributes(self):
+        self._record(self._lease(self.SID_A, 703, OLD_HEAD))
+        self.assertIsNone(
+            leases.load_session_lease_shadow(session_id=self.SID_B)
+        )
+        verdict = leases.assess_crashed_session_lease_orphan(
+            leases.load_session_lease_shadow(session_id=self.SID_A),
+            lease={"session_id": self.SID_B},
+            pid_alive=False,
+        )
+        self.assertFalse(verdict["orphan_evidence"])
+
+    def test_clearing_one_session_preserves_the_other(self):
+        self._record(self._lease(self.SID_A, 703, OLD_HEAD))
+        self._record(self._lease(self.SID_B, 701, NEW_HEAD))
+        # Sanctioned clear of B (the currently recorded in-memory lease).
+        leases.clear_session_lease()
+        self.assertIsNone(
+            leases.load_session_lease_shadow(session_id=self.SID_B)
+        )
+        shadow_a = leases.load_session_lease_shadow(session_id=self.SID_A)
+        self.assertIsNotNone(shadow_a)
+        self.assertEqual(shadow_a.get("session_id"), self.SID_A)
+
+    def test_reconnected_process_recovers_by_session_id(self):
+        self._record(self._lease(self.SID_A, 703, OLD_HEAD))
+        self._record(self._lease(self.SID_B, 701, NEW_HEAD))
+        # Simulated crash + reconnect: in-memory state is gone, durable
+        # records remain enumerable and individually attributable.
+        leases._SESSION_LEASE = None
+        records = mss.list_states(kind=mss.KIND_REVIEWER_SESSION_LEASE)
+        self.assertEqual(
+            sorted(r.get("session_id") for r in records),
+            sorted([self.SID_A, self.SID_B]),
+        )
+        shadow = leases.load_session_lease_shadow(session_id=self.SID_B)
+        self.assertEqual(shadow.get("pr_number"), 701)
+
+    def test_legacy_single_slot_record_still_resolves_by_session_id(self):
+        # Upgrade path: a record written by pre-F5 code (no instance key)
+        # remains usable when its payload names the requested session.
+        mss.save_state(
+            kind=mss.KIND_REVIEWER_SESSION_LEASE,
+            payload={"session_id": self.SID_A, "owner_pid": os.getpid()},
+        )
+        shadow = leases.load_session_lease_shadow(session_id=self.SID_A)
+        self.assertIsNotNone(shadow)
+        self.assertIsNone(
+            leases.load_session_lease_shadow(session_id=self.SID_B)
+        )
+
+
+class TestF6AuditBeforeClear(_CapabilityHarness):
+    """F6: the durable audit record is persisted before the env clears."""
+
+    def _recovery_env(self, missing: str) -> dict:
+        return self._env(
+            GITEA_ACTIVE_WORKTREE=missing,
+            GITEA_FORCE_STALE_BINDING_RECOVERY="1",
+        )
+
+    def test_audit_write_failure_blocks_the_clear(self):
+        missing = self._deleted_worktree()
+        with patch.dict(os.environ, self._recovery_env(missing)):
+            with patch.object(
+                mss, "save_state", side_effect=OSError("disk full")
+            ):
+                report = mcp_server._assess_stale_active_binding(
+                    auto_recover=True
+                )
+            self.assertEqual(os.environ.get("GITEA_ACTIVE_WORKTREE"), missing)
+        self.assertFalse(report["recovery_performed"])
+        self.assertIs(report.get("audit_persisted"), False)
+        self.assertTrue(report.get("audit_write_failed"))
+        self.assertIn("NOT cleared", " ".join(report.get("reasons") or []))
+
+    def test_retry_after_audit_failure_recovers(self):
+        missing = self._deleted_worktree()
+        with patch.dict(os.environ, self._recovery_env(missing)):
+            with patch.object(
+                mss, "save_state", side_effect=OSError("disk full")
+            ):
+                first = mcp_server._assess_stale_active_binding(
+                    auto_recover=True
+                )
+            self.assertFalse(first["recovery_performed"])
+            # Persistence recovered; the retry clears with a durable audit.
+            second = mcp_server._assess_stale_active_binding(auto_recover=True)
+            self.assertTrue(second["recovery_performed"])
+            self.assertIs(second.get("audit_persisted"), True)
+            self.assertNotIn("GITEA_ACTIVE_WORKTREE", os.environ)
+            audit = mss.load_state(kind=mss.KIND_STALE_BINDING_RECOVERY)
+        self.assertIsNotNone(audit)
+        self.assertEqual(audit.get("recovery_status"), "performed")
+        self.assertEqual(audit.get("cleared_value"), missing)
+
+    def test_audit_record_exists_before_env_clear(self):
+        missing = self._deleted_worktree()
+        observed: list[tuple[str | None, str | None]] = []
+        real_save = mss.save_state
+
+        def _spy(**kwargs):
+            observed.append(
+                (
+                    (kwargs.get("payload") or {}).get("recovery_status"),
+                    os.environ.get("GITEA_ACTIVE_WORKTREE"),
+                )
+            )
+            return real_save(**kwargs)
+
+        with patch.dict(os.environ, self._recovery_env(missing)):
+            with patch.object(mss, "save_state", side_effect=_spy):
+                report = mcp_server._assess_stale_active_binding(
+                    auto_recover=True
+                )
+        self.assertTrue(report["recovery_performed"])
+        self.assertGreaterEqual(len(observed), 2)
+        pending_status, env_at_pending = observed[0]
+        performed_status, env_at_performed = observed[-1]
+        # Ordering proof: the binding was still present while the pending
+        # audit persisted, and gone by the time the performed status wrote.
+        self.assertEqual(pending_status, "pending")
+        self.assertEqual(env_at_pending, missing)
+        self.assertEqual(performed_status, "performed")
+        self.assertIsNone(env_at_performed)
+
+    def test_recovery_is_idempotent_after_success(self):
+        missing = self._deleted_worktree()
+        with patch.dict(os.environ, self._recovery_env(missing)):
+            first = mcp_server._assess_stale_active_binding(auto_recover=True)
+            second = mcp_server._assess_stale_active_binding(auto_recover=True)
+        self.assertTrue(first["recovery_performed"])
+        self.assertEqual(second["classification"], sbr.CLASSIFICATION_UNBOUND)
+        self.assertFalse(second["recovery_performed"])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_issue_702_stale_binding_lease_recovery.py

@@ -0,0 +1,517 @@
+"""Regression tests for #702: stale runtime binding + orphaned durable lease.
+
+Reproduces the PR #701 incident (lease 65664-4f248d213d60, comments
+11129/11131): the MCP daemon crashed without teardown, destroying the
+in-memory session lease while the durable comment lease survived, and the
+auto-reconnected daemon inherited a stale ``GITEA_ACTIVE_WORKTREE`` binding
+(``branches/review-pr-654``) from its parent environment.
+
+Covers the five mandated scenarios:
+
+1. lost in-session leases  — durable shadow survives a crash and yields
+   provable owner-process evidence
+2. old-head leases         — expired superseded-head lease with no terminal
+   review becomes recoverable only with orphan evidence + controller authority
+3. runtime/worktree mismatch — env bindings to deleted worktrees are demoted,
+   never silently bound
+4. managed reconnect       — boot-inherited uncorroborated bindings are
+   surfaced for managed recovery; provably stale ones are clear-eligible
+5. safe expiry             — pre-expiry stays fail-closed wait (no steal);
+   canonical expiry unlocks acquisition and guarded cleanup
+"""
+
+from __future__ import annotations
+
+import os
+import sys
+import unittest
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+import namespace_workspace_binding as nwb  # noqa: E402
+import reviewer_pr_lease as leases  # noqa: E402
+import stale_binding_recovery as sbr  # noqa: E402
+
+# PR #701 incident fixtures.
+OLD_HEAD = "b4d0cb22e452a07c8e816745c704ddb0f43edf0b"
+NEW_HEAD = "6b675f5c834b41f9d74e8a54294ff44dddf28ae4"
+CRASHED_SESSION = "65664-4f248d213d60"
+LEASE_COMMENT = 11131
+REPO = "Scaled-Tech-Consulting/Gitea-Tools"
+PR = 701
+ISSUE = 699
+LEASE_WORKTREE = "branches/review-fix-issue-699-structured-auth-mcp-errors"
+
+
+def _now() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+def _lease_comment(
+    *,
+    phase: str = "validation-start",
+    candidate_head: str = OLD_HEAD,
+    session_id: str = CRASHED_SESSION,
+    comment_id: int = LEASE_COMMENT,
+    last_activity: datetime | None = None,
+    expires_at: datetime | None = None,
+    worktree: str = LEASE_WORKTREE,
+) -> dict:
+    stamp = last_activity or _now()
+    body = leases.format_lease_body(
+        repo=REPO,
+        pr_number=PR,
+        issue_number=ISSUE,
+        reviewer_identity="sysadmin",
+        profile="prgs-reviewer",
+        session_id=session_id,
+        worktree=worktree,
+        phase=phase,
+        candidate_head=candidate_head,
+        target_branch="master",
+        target_branch_sha="2" * 40,
+        last_activity=stamp,
+        expires_at=expires_at,
+    )
+    return {
+        "id": comment_id,
+        "body": body,
+        "user": {"login": "sysadmin"},
+        "author": "sysadmin",
+        "created_at": stamp.isoformat(),
+        "updated_at": stamp.isoformat(),
+    }
+
+
+def _expired_superseded_comments() -> list[dict]:
+    stale = _now() - timedelta(hours=3)
+    return [_lease_comment(
+        last_activity=stale,
+        expires_at=stale + timedelta(minutes=120),
+    )]
+
+
+def _fresh_superseded_comments() -> list[dict]:
+    recent = _now() - timedelta(minutes=10)
+    return [_lease_comment(
+        last_activity=recent,
+        expires_at=recent + timedelta(minutes=120),
+    )]
+
+
+class TestLostInSessionLeaseShadow(unittest.TestCase):
+    """Scenario 1: the durable shadow survives a daemon crash (#702 AC3)."""
+
+    def setUp(self):
+        leases._SESSION_LEASE = None
+
+    def tearDown(self):
+        leases._SESSION_LEASE = None
+
+    def _record(self) -> dict:
+        return leases.record_session_lease(
+            {
+                "pr_number": PR,
+                "issue_number": ISSUE,
+                "session_id": CRASHED_SESSION,
+                "reviewer_identity": "sysadmin",
+                "profile": "prgs-reviewer",
+                "worktree": LEASE_WORKTREE,
+                "phase": "claimed",
+                "candidate_head": OLD_HEAD,
+                "repo": REPO,
+                "comment_id": LEASE_COMMENT,
+            },
+            lease_provenance={"source": "acquire", "comment_id": LEASE_COMMENT},
+        )
+
+    def test_shadow_written_on_record_and_survives_simulated_crash(self):
+        self._record()
+        # Simulated unhandled crash: the in-memory dict dies with the process
+        # but the sanctioned clear path never runs.
+        leases._SESSION_LEASE = None
+        shadow = leases.load_session_lease_shadow()
+        self.assertIsNotNone(shadow)
+        self.assertEqual(shadow.get("session_id"), CRASHED_SESSION)
+        self.assertEqual(shadow.get("pr_number"), PR)
+        self.assertEqual(int(shadow.get("owner_pid")), os.getpid())
+
+    def test_sanctioned_clear_removes_shadow(self):
+        self._record()
+        leases.clear_session_lease()
+        self.assertIsNone(leases.load_session_lease_shadow())
+
+    def test_orphan_assessment_dead_owner_pid_proves_exit(self):
+        self._record()
+        leases._SESSION_LEASE = None
+        shadow = leases.load_session_lease_shadow()
+        lease = {"session_id": CRASHED_SESSION}
+        verdict = leases.assess_crashed_session_lease_orphan(
+            shadow, lease=lease, current_pid=os.getpid() + 1, pid_alive=False
+        )
+        self.assertTrue(verdict["orphan_evidence"])
+        self.assertIs(verdict["owner_process_alive"], False)
+
+    def test_orphan_assessment_alive_pid_is_not_ownership_proof(self):
+        self._record()
+        shadow = leases.load_session_lease_shadow()
+        verdict = leases.assess_crashed_session_lease_orphan(
+            shadow,
+            lease={"session_id": CRASHED_SESSION},
+            current_pid=os.getpid() + 1,
+            pid_alive=True,
+        )
+        self.assertFalse(verdict["orphan_evidence"])
+        self.assertIsNone(verdict["owner_process_alive"])
+
+    def test_orphan_assessment_session_mismatch_proves_nothing(self):
+        self._record()
+        shadow = leases.load_session_lease_shadow()
+        verdict = leases.assess_crashed_session_lease_orphan(
+            shadow, lease={"session_id": "other-999"}, pid_alive=False
+        )
+        self.assertFalse(verdict["orphan_evidence"])
+        self.assertIsNone(verdict["owner_process_alive"])
+
+    def test_orphan_assessment_without_shadow_is_unknown(self):
+        verdict = leases.assess_crashed_session_lease_orphan(
+            None, lease={"session_id": CRASHED_SESSION}
+        )
+        self.assertFalse(verdict["orphan_evidence"])
+        self.assertIsNone(verdict["owner_process_alive"])
+
+
+class TestOldHeadLeaseCleanup(unittest.TestCase):
+    """Scenario 2: expired superseded-head lease with no terminal review."""
+
+    def _assess(self, comments, **kwargs):
+        defaults = dict(
+            pr_number=PR,
+            current_head_sha=NEW_HEAD,
+            formal_reviews=[],
+            repo=REPO,
+            expected_repo=REPO,
+            requesting_session_id="fresh-controller",
+            controller_recovery_authorized=True,
+            worktree_exists=True,
+            worktree_clean=True,
+            owner_process_alive=False,
+        )
+        defaults.update(kwargs)
+        return leases.assess_obsolete_reviewer_comment_lease_cleanup(
+            comments, **defaults
+        )
+
+    def test_expired_orphan_with_full_evidence_is_cleanup_eligible(self):
+        result = self._assess(_expired_superseded_comments())
+        self.assertEqual(result["classification"], "orphaned_expired_superseded_head")
+        self.assertTrue(result["cleanup_allowed"])
+        self.assertIn("phase: released", result["release_body"] or "")
+        self.assertEqual(
+            result["exact_next_action"], "cleanup_obsolete_reviewer_comment_lease"
+        )
+
+    def test_expired_orphan_without_owner_evidence_fails_closed(self):
+        result = self._assess(
+            _expired_superseded_comments(), owner_process_alive=None
+        )
+        self.assertFalse(result["cleanup_allowed"])
+        self.assertIn(
+            "provable owner-process-exit evidence",
+            " ".join(result["fail_closed_reasons"]),
+        )
+
+    def test_expired_orphan_without_controller_authority_fails_closed(self):
+        result = self._assess(
+            _expired_superseded_comments(), controller_recovery_authorized=False
+        )
+        self.assertFalse(result["cleanup_allowed"])
+
+    def test_expired_orphan_with_dirty_worktree_fails_closed(self):
+        result = self._assess(
+            _expired_superseded_comments(),
+            worktree_clean=False,
+            worktree_has_unpreserved_work=True,
+        )
+        self.assertFalse(result["cleanup_allowed"])
+
+    def test_unexpired_superseded_no_terminal_stays_ambiguous_wait(self):
+        # Regression guard: pre-expiry there is still no steal path.
+        result = self._assess(_fresh_superseded_comments())
+        self.assertEqual(result["classification"], "ambiguous_conflicting_evidence")
+        self.assertFalse(result["cleanup_allowed"])
+        self.assertEqual(result["exact_next_action"], "wait")
+
+
+class TestRuntimeWorktreeMismatch(unittest.TestCase):
+    """Scenario 3: env bindings to deleted worktrees are demoted (#702 AC2)."""
+
+    def test_missing_active_env_binding_is_demoted(self):
+        demotions: list[str] = []
+        missing = "/nonexistent/branches/review-pr-654"
+        path, source = nwb.resolve_namespace_workspace(
+            role_kind="reviewer",
+            process_project_root=os.getcwd(),
+            env={"GITEA_ACTIVE_WORKTREE": missing},
+            demotions=demotions,
+            verify_paths=True,
+        )
+        self.assertEqual(source, "MCP server process root (default)")
+        self.assertEqual(len(demotions), 1)
+        self.assertIn("no longer exists", demotions[0])
+
+    def test_missing_active_falls_through_to_existing_role_env(self):
+        existing = os.getcwd()
+        path, source = nwb.resolve_namespace_workspace(
+            role_kind="reviewer",
+            process_project_root="/tmp",
+            env={
+                "GITEA_ACTIVE_WORKTREE": "/nonexistent/branches/review-pr-654",
+                "GITEA_REVIEWER_WORKTREE": existing,
+            },
+            verify_paths=True,
+        )
+        self.assertEqual(path, os.path.realpath(existing))
+        self.assertEqual(source, "GITEA_REVIEWER_WORKTREE environment variable")
+
+    def test_existing_active_env_binding_still_wins(self):
+        existing = os.getcwd()
+        path, source = nwb.resolve_namespace_workspace(
+            role_kind="reviewer",
+            process_project_root="/tmp",
+            env={"GITEA_ACTIVE_WORKTREE": existing},
+            verify_paths=True,
+        )
+        self.assertEqual(path, os.path.realpath(existing))
+        self.assertEqual(source, "GITEA_ACTIVE_WORKTREE environment variable")
+
+    def test_explicit_worktree_path_argument_is_never_demoted(self):
+        missing = "/nonexistent/branches/explicit"
+        path, source = nwb.resolve_namespace_workspace(
+            role_kind="reviewer",
+            worktree_path=missing,
+            process_project_root="/tmp",
+            env={},
+            verify_paths=True,
+        )
+        self.assertEqual(source, "worktree_path argument")
+
+    def test_mutation_context_reports_demotion_in_ignored_bindings(self):
+        ctx = nwb.resolve_namespace_mutation_context(
+            role_kind="reviewer",
+            worktree_path=None,
+            process_project_root=os.getcwd(),
+            env={"GITEA_ACTIVE_WORKTREE": "/nonexistent/branches/review-pr-654"},
+        )
+        joined = " ".join(ctx["ignored_bindings"])
+        self.assertIn("stale binding, #702", joined)
+
+
+class TestManagedReconnectRecovery(unittest.TestCase):
+    """Scenario 4: boot-inherited bindings and the sanctioned recovery plan."""
+
+    def test_uncorroborated_inherited_reviewer_binding_is_unverified(self):
+        classification = sbr.classify_active_worktree_binding(
+            active_value=os.getcwd(),
+            role_env_value=None,
+            session_lease_worktree=None,
+            boot_inherited=True,
+            path_exists=True,
+            role_kind="reviewer",
+        )
+        self.assertEqual(
+            classification["classification"], sbr.CLASSIFICATION_UNVERIFIED_INHERITED
+        )
+        self.assertFalse(classification["clear_eligible"])
+        plan = sbr.plan_recovery(classification)
+        self.assertFalse(plan["clear_allowed"])
+        self.assertIn("managed", plan["exact_next_action"])
+        self.assertIn("do not clear or rewrite", plan["exact_next_action"])
+
+    def test_missing_path_binding_is_provably_stale_and_clear_eligible(self):
+        classification = sbr.classify_active_worktree_binding(
+            active_value="/nonexistent/branches/review-pr-654",
+            boot_inherited=True,
+            path_exists=False,
+            role_kind="reviewer",
+        )
+        self.assertEqual(
+            classification["classification"],
+            sbr.CLASSIFICATION_PROVABLY_STALE_MISSING_PATH,
+        )
+        plan = sbr.plan_recovery(classification)
+        self.assertTrue(plan["clear_allowed"])
+
+    def test_inherited_binding_superseded_by_session_lease_is_clear_eligible(self):
+        classification = sbr.classify_active_worktree_binding(
+            active_value="/tmp",
+            session_lease_worktree=os.getcwd(),
+            boot_inherited=True,
+            path_exists=True,
+            role_kind="reviewer",
+        )
+        self.assertEqual(
+            classification["classification"],
+            sbr.CLASSIFICATION_SUPERSEDED_BY_SESSION_LEASE,
+        )
+        self.assertTrue(sbr.plan_recovery(classification)["clear_allowed"])
+
+    def test_post_boot_binding_conflict_is_surfaced_not_cleared(self):
+        classification = sbr.classify_active_worktree_binding(
+            active_value="/tmp",
+            session_lease_worktree=os.getcwd(),
+            boot_inherited=False,
+            path_exists=True,
+            role_kind="reviewer",
+        )
+        self.assertEqual(
+            classification["classification"], sbr.CLASSIFICATION_UNVERIFIED_INHERITED
+        )
+        self.assertFalse(sbr.plan_recovery(classification)["clear_allowed"])
+
+    def test_corroborated_bindings_are_untouched(self):
+        for kwargs in (
+            dict(role_env_value=os.getcwd()),
+            dict(session_lease_worktree=os.getcwd()),
+        ):
+            classification = sbr.classify_active_worktree_binding(
+                active_value=os.getcwd(),
+                boot_inherited=True,
+                path_exists=True,
+                role_kind="reviewer",
+                **kwargs,
+            )
+            self.assertEqual(
+                classification["classification"], sbr.CLASSIFICATION_CORROBORATED
+            )
+            self.assertFalse(sbr.plan_recovery(classification)["clear_allowed"])
+
+    def test_author_inherited_binding_stays_corroborated(self):
+        classification = sbr.classify_active_worktree_binding(
+            active_value=os.getcwd(),
+            boot_inherited=True,
+            path_exists=True,
+            role_kind="author",
+        )
+        self.assertEqual(
+            classification["classification"], sbr.CLASSIFICATION_CORROBORATED
+        )
+
+    def test_apply_recovery_clears_env_only_for_authorized_plan(self):
+        env = {"GITEA_ACTIVE_WORKTREE": "/nonexistent/branches/review-pr-654"}
+        plan = sbr.plan_recovery(
+            sbr.classify_active_worktree_binding(
+                active_value=env["GITEA_ACTIVE_WORKTREE"],
+                boot_inherited=True,
+                path_exists=False,
+                role_kind="reviewer",
+            )
+        )
+        applied = sbr.apply_recovery(plan, env=env)
+        self.assertTrue(applied["performed"])
+        self.assertNotIn("GITEA_ACTIVE_WORKTREE", env)
+        self.assertIn("review-pr-654", applied["cleared_value"])
+
+    def test_apply_recovery_refuses_unauthorized_plan(self):
+        env = {"GITEA_ACTIVE_WORKTREE": os.getcwd()}
+        plan = sbr.plan_recovery(
+            sbr.classify_active_worktree_binding(
+                active_value=env["GITEA_ACTIVE_WORKTREE"],
+                boot_inherited=True,
+                path_exists=True,
+                role_kind="reviewer",
+            )
+        )
+        applied = sbr.apply_recovery(plan, env=env)
+        self.assertFalse(applied["performed"])
+        self.assertIn("GITEA_ACTIVE_WORKTREE", env)
+
+
+class TestSafeExpiryDiagnosis(unittest.TestCase):
+    """Scenario 5: canonical expiry flips wait into acquire/guarded cleanup."""
+
+    def _diagnose(self, comments, **kwargs):
+        defaults = dict(
+            pr_number=PR,
+            current_session_id=None,
+            current_reviewer_identity="fresh-reviewer",
+            current_head_sha=NEW_HEAD,
+            formal_reviews=[],
+        )
+        defaults.update(kwargs)
+        return leases.diagnose_reviewer_pr_lease_handoff(comments, **defaults)
+
+    def setUp(self):
+        leases._SESSION_LEASE = None
+
+    def test_pre_expiry_superseded_no_terminal_stays_wait(self):
+        diagnosis = self._diagnose(_fresh_superseded_comments())
+        self.assertEqual(
+            diagnosis["classification"], "ambiguous_conflicting_evidence"
+        )
+        self.assertEqual(diagnosis["next_action"], "wait")
+
+    def test_post_expiry_without_orphan_evidence_unlocks_acquire(self):
+        diagnosis = self._diagnose(_expired_superseded_comments())
+        self.assertEqual(
+            diagnosis["classification"], "orphaned_expired_superseded_head"
+        )
+        self.assertEqual(diagnosis["next_action"], "acquire")
+
+    def test_post_expiry_with_orphan_evidence_offers_guarded_cleanup(self):
+        diagnosis = self._diagnose(
+            _expired_superseded_comments(),
+            owner_process_alive=False,
+            worktree_clean=True,
+            worktree_exists=True,
+        )
+        self.assertEqual(
+            diagnosis["classification"], "orphaned_expired_superseded_head"
+        )
+        self.assertEqual(
+            diagnosis["next_action"], "cleanup_obsolete_reviewer_comment_lease"
+        )
+
+    def test_expired_lease_no_longer_blocks_fresh_acquisition(self):
+        assessment = leases.assess_acquire_lease(
+            _expired_superseded_comments(),
+            pr_number=PR,
+            reviewer_identity="fresh-reviewer",
+            profile="prgs-reviewer",
+            session_id="99999-fresh",
+            repo=REPO,
+            issue_number=ISSUE,
+            worktree="branches/review-pr-701-fresh",
+            candidate_head=NEW_HEAD,
+            target_branch="master",
+            target_branch_sha="2" * 40,
+        )
+        self.assertTrue(assessment.get("acquire_allowed"))
+
+    def test_unparseable_expiry_fails_closed_in_cleanup(self):
+        comment = _lease_comment(
+            last_activity=_now() - timedelta(hours=3), expires_at=None
+        )
+        comment["body"] = comment["body"].replace(
+            "expires_at: ", "expires_at: not-a-timestamp"
+        )
+        result = leases.assess_obsolete_reviewer_comment_lease_cleanup(
+            [comment],
+            pr_number=PR,
+            current_head_sha=NEW_HEAD,
+            formal_reviews=[],
+            repo=REPO,
+            expected_repo=REPO,
+            requesting_session_id="fresh-controller",
+            controller_recovery_authorized=True,
+            worktree_exists=True,
+            worktree_clean=True,
+            owner_process_alive=False,
+        )
+        self.assertFalse(result["cleanup_allowed"])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_issue_723_role_stamp_and_submission.py

@@ -1,337 +0,0 @@
-"""#723 AC3/AC4/AC5: denied resolves must not poison the session role stamp,
-review-submission workspace failures must fail closed with structured
-reasons, and runtime-context capabilities must apply the resolver's
-role-exclusive filter.
-
-Reproduces the PR #721 sequence: a reviewer session resolved a task whose
-capability-map role had drifted to ``merger``; the denied resolve stamped
-``_preflight_resolved_role = "merger"`` anyway, ``mark_final`` succeeded, and
-``gitea_submit_pr_review`` raised RuntimeError out of the tool as a generic
-``internal_error``.
-"""
-
-from __future__ import annotations
-
-import os
-import sys
-import unittest
-from pathlib import Path
-from unittest.mock import patch
-
-ROOT = str(Path(__file__).resolve().parent.parent)
-if ROOT not in sys.path:
-    sys.path.insert(0, ROOT)
-
-import gitea_mcp_server as mcp_server
-import task_capability_map
-
-
-REVIEWER_PROFILE = {
-    "profile_name": "prgs-reviewer",
-    "role": "reviewer",
-    "allowed_operations": [
-        "gitea.read",
-        "gitea.pr.review",
-        "gitea.pr.approve",
-        "gitea.pr.request_changes",
-        "gitea.pr.comment",
-        "gitea.issue.comment",
-    ],
-    "forbidden_operations": [
-        "gitea.branch.create",
-        "gitea.branch.push",
-        "gitea.repo.commit",
-        "gitea.pr.create",
-        "gitea.pr.merge",
-    ],
-}
-
-CONFIG = {
-    "profiles": {
-        "prgs-reviewer": {
-            "role": "reviewer",
-            "allowed_operations": REVIEWER_PROFILE["allowed_operations"],
-            "forbidden_operations": REVIEWER_PROFILE["forbidden_operations"],
-        },
-        "prgs-merger": {
-            "role": "merger",
-            "allowed_operations": [
-                "gitea.read",
-                "gitea.pr.merge",
-                "gitea.pr.comment",
-                "gitea.issue.comment",
-            ],
-            "forbidden_operations": [
-                "gitea.pr.approve",
-                "gitea.pr.review",
-                "gitea.pr.request_changes",
-            ],
-        },
-    }
-}
-
-
-def _reset_preflight():
-    mcp_server._clear_preflight_capability_state()
-    mcp_server._preflight_whoami_called = False
-    mcp_server._preflight_whoami_violation = False
-
-
-class _ResolveHarness(unittest.TestCase):
-    """Shared patched-resolver harness."""
-
-    def setUp(self):
-        _reset_preflight()
-
-    def tearDown(self):
-        _reset_preflight()
-
-    def _resolve(self, task, profile=REVIEWER_PROFILE, required_role=None):
-        """Run gitea_resolve_task_capability with a fixed profile/config."""
-        patches = [
-            patch.object(mcp_server, "get_profile", return_value=profile),
-            patch.object(
-                mcp_server.gitea_config, "load_config", return_value=CONFIG
-            ),
-            patch.object(
-                mcp_server, "_authenticated_username", return_value="tester"
-            ),
-            patch.object(
-                mcp_server, "_ensure_matching_profile", return_value=None
-            ),
-            patch.object(
-                mcp_server, "init_review_decision_lock", return_value=None
-            ),
-        ]
-        if required_role is not None:
-            patches.append(
-                patch.object(
-                    mcp_server.task_capability_map,
-                    "required_role",
-                    side_effect=lambda t: (
-                        required_role
-                        if t == task
-                        else task_capability_map.TASK_CAPABILITY_MAP[t]["role"]
-                    ),
-                )
-            )
-        for p in patches:
-            p.__enter__()
-        try:
-            return mcp_server.gitea_resolve_task_capability(
-                task=task, remote="prgs"
-            )
-        finally:
-            for p in reversed(patches):
-                p.__exit__(None, None, None)
-
-
-class TestAC3DeniedResolveDoesNotStampRole(_ResolveHarness):
-    def test_allowed_resolve_stamps_role(self):
-        result = self._resolve("review_pr")
-        self.assertTrue(result["allowed_in_current_session"], result)
-        self.assertEqual(mcp_server._preflight_resolved_role, "reviewer")
-        self.assertEqual(mcp_server._preflight_resolved_task, "review_pr")
-
-    def test_denied_resolve_does_not_stamp_required_role(self):
-        """The PR #721 poison: review_pr requiring merger under a reviewer."""
-        result = self._resolve("review_pr", required_role="merger")
-        self.assertFalse(result["allowed_in_current_session"], result)
-        self.assertIsNone(
-            mcp_server._preflight_resolved_role,
-            "denied resolve must not stamp the required role (#723 AC3)",
-        )
-        self.assertIsNone(mcp_server._preflight_resolved_task)
-
-    def test_denied_resolve_clears_prior_stale_stamp(self):
-        allowed = self._resolve("review_pr")
-        self.assertTrue(allowed["allowed_in_current_session"])
-        self.assertEqual(mcp_server._preflight_resolved_role, "reviewer")
-
-        denied = self._resolve("merge_pr")  # reviewer profile cannot merge
-        self.assertFalse(denied["allowed_in_current_session"], denied)
-        self.assertIsNone(
-            mcp_server._preflight_resolved_role,
-            "a denied resolve must clear the previous stamp, not keep it",
-        )
-
-    def test_denied_resolve_keeps_effective_role_on_actual_profile(self):
-        with patch.object(
-            mcp_server, "get_profile", return_value=REVIEWER_PROFILE
-        ):
-            self._resolve("review_pr", required_role="merger")
-            self.assertEqual(
-                mcp_server._effective_workspace_role(), "reviewer"
-            )
-
-
-class TestAC4SubmissionFailsClosedStructured(unittest.TestCase):
-    def setUp(self):
-        _reset_preflight()
-
-    def tearDown(self):
-        _reset_preflight()
-
-    def test_workspace_binding_failure_returns_reasons_not_raise(self):
-        boom = RuntimeError(
-            "namespace workspace binding blocked: role 'merger' cannot "
-            "mutate from reviewer session workspace"
-        )
-        with patch.object(
-            mcp_server,
-            "_verify_role_mutation_workspace",
-            side_effect=boom,
-        ), patch.object(
-            mcp_server, "get_profile", return_value=REVIEWER_PROFILE
-        ):
-            result = mcp_server._evaluate_pr_review_submission(
-                pr_number=721,
-                action="approve",
-                expected_head_sha="80f59b334e6671b08006725292c08a8e8b6c823f",
-                remote="prgs",
-                live=True,
-                final_review_decision_ready=True,
-            )
-        self.assertFalse(result["performed"])
-        self.assertEqual(result["blocker_kind"], "workspace_role_binding")
-        self.assertTrue(
-            any("workspace/role binding failed" in r for r in result["reasons"]),
-            result["reasons"],
-        )
-        self.assertTrue(
-            any("cannot mutate from reviewer session" in r for r in result["reasons"]),
-            "the underlying binding error text must be preserved",
-        )
-
-    def test_stale_runtime_failure_also_structured(self):
-        boom = RuntimeError(
-            "stale-runtime: The active Gitea MCP server process is stale"
-        )
-        with patch.object(
-            mcp_server,
-            "_verify_role_mutation_workspace",
-            side_effect=boom,
-        ), patch.object(
-            mcp_server, "get_profile", return_value=REVIEWER_PROFILE
-        ):
-            result = mcp_server._evaluate_pr_review_submission(
-                pr_number=721,
-                action="approve",
-                remote="prgs",
-                live=True,
-            )
-        self.assertFalse(result["performed"])
-        self.assertEqual(result["blocker_kind"], "workspace_role_binding")
-        self.assertTrue(
-            any("stale-runtime" in r for r in result["reasons"]),
-            result["reasons"],
-        )
-
-    def test_dry_run_also_fails_closed_structured(self):
-        boom = RuntimeError("binding blocked")
-        with patch.object(
-            mcp_server,
-            "_verify_role_mutation_workspace",
-            side_effect=boom,
-        ), patch.object(
-            mcp_server, "get_profile", return_value=REVIEWER_PROFILE
-        ):
-            result = mcp_server._evaluate_pr_review_submission(
-                pr_number=721,
-                action="approve",
-                remote="prgs",
-                live=False,
-            )
-        self.assertFalse(result["would_perform"])
-        self.assertEqual(result["blocker_kind"], "workspace_role_binding")
-
-
-class TestAC5RuntimeContextRoleFilter(unittest.TestCase):
-    def test_role_filtered_view_matches_resolver_denial(self):
-        """Reviewer session: merge_pr stays denied under the role filter even
-        when the permission is (pathologically) present."""
-        allowed_ops = [
-            "gitea.read",
-            "gitea.pr.review",
-            "gitea.pr.approve",
-            "gitea.pr.request_changes",
-            "gitea.pr.comment",
-            "gitea.issue.comment",
-            "gitea.pr.merge",
-        ]
-        caps = mcp_server._build_runtime_task_capabilities(
-            allowed_ops, [], CONFIG, active_role_kind="reviewer"
-        )
-        merge_entry = next(
-            t for t in caps["task_capabilities"] if t["task"] == "merge_pr"
-        )
-        self.assertTrue(merge_entry["role_exclusive"])
-        self.assertEqual(merge_entry["capability_view"], "role_filtered")
-        self.assertFalse(
-            merge_entry["allowed_in_current_session"],
-            "role-exclusive merge_pr must stay denied for a reviewer role "
-            "even when the permission is present (#723 AC5)",
-        )
-        self.assertFalse(caps["can_merge_prs"])
-
-    def test_role_filter_restricts_matching_profiles(self):
-        caps = mcp_server._build_runtime_task_capabilities(
-            ["gitea.read"], [], CONFIG, active_role_kind="author"
-        )
-        review_entry = next(
-            t for t in caps["task_capabilities"] if t["task"] == "review_pr"
-        )
-        self.assertEqual(
-            review_entry["matching_configured_profiles"],
-            ["prgs-reviewer"],
-            "role-exclusive review_pr must not list the merger profile",
-        )
-        merge_entry = next(
-            t for t in caps["task_capabilities"] if t["task"] == "merge_pr"
-        )
-        self.assertEqual(
-            merge_entry["matching_configured_profiles"], ["prgs-merger"]
-        )
-
-    def test_permission_only_view_is_labeled(self):
-        caps = mcp_server._build_runtime_task_capabilities(
-            ["gitea.pr.merge", "gitea.read"], [], CONFIG
-        )
-        merge_entry = next(
-            t for t in caps["task_capabilities"] if t["task"] == "merge_pr"
-        )
-        self.assertEqual(merge_entry["capability_view"], "permission_only")
-        # Legacy permission-only semantics preserved when no role supplied.
-        self.assertTrue(merge_entry["allowed_in_current_session"])
-
-    def test_incident_722_shape_runtime_context_agrees_with_resolver(self):
-        """Post-970e68b shape: review_pr requires merger; a reviewer session
-        must see review_pr denied in runtime context, matching the resolver."""
-        with patch.object(
-            mcp_server.task_capability_map,
-            "required_role",
-            side_effect=lambda t: (
-                "merger"
-                if t == "review_pr"
-                else task_capability_map.TASK_CAPABILITY_MAP[t]["role"]
-            ),
-        ):
-            caps = mcp_server._build_runtime_task_capabilities(
-                REVIEWER_PROFILE["allowed_operations"],
-                REVIEWER_PROFILE["forbidden_operations"],
-                CONFIG,
-                active_role_kind="reviewer",
-            )
-        review_entry = next(
-            t for t in caps["task_capabilities"] if t["task"] == "review_pr"
-        )
-        self.assertFalse(
-            review_entry["allowed_in_current_session"],
-            "runtime context must not report review_pr allowed when the "
-            "resolver would fail-close it (incident #722)",
-        )
-        self.assertFalse(caps["can_review_prs"])
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/test_mcp_server.py

@@ -2567,15 +2567,18 @@ class TestSubmitPrReview(unittest.TestCase):
         lock = _load_review_decision_lock() or {}
         lock["live_mutations"] = [{"pr_number": 8, "action": "approve", "review_id": 42, "review_state": "approve"}]
         _save_review_decision_lock(lock)
-        r = gitea_authorize_review_correction(
-            prior_review_id=42,
-            prior_review_state="approve",
-            reason="typo",
-            operator_authorized=True,
-        )
+        with patch("mcp_server.api_request", return_value={"id": 9001}):
+            r = gitea_authorize_review_correction(
+                prior_review_id=42,
+                prior_review_state="approve",
+                reason="typo",
+                operator_authorized=True,
+                target_pr_number=8,
+            )
         self.assertTrue(r["authorized"])
         lock = _load_review_decision_lock()
         self.assertTrue(lock["correction_authorized"])
+        self.assertEqual(lock["correction_pr_number"], 8)
 
     def test_authorize_review_correction_mismatch(self):
         from mcp_server import _load_review_decision_lock, _save_review_decision_lock
@@ -2589,6 +2592,7 @@ class TestSubmitPrReview(unittest.TestCase):
             prior_review_state="approve",
             reason="typo",
             operator_authorized=True,
+            target_pr_number=8,
         )
         self.assertFalse(r["authorized"])
         self.assertTrue(any("prior review ID" in x for x in r["reasons"]))
@@ -2599,10 +2603,22 @@ class TestSubmitPrReview(unittest.TestCase):
             prior_review_state="request_changes",
             reason="typo",
             operator_authorized=True,
+            target_pr_number=8,
         )
         self.assertFalse(r["authorized"])
         self.assertTrue(any("prior review state" in x for x in r["reasons"]))
 
+        # Cross-PR unlock refused (#693)
+        r = gitea_authorize_review_correction(
+            prior_review_id=42,
+            prior_review_state="approve",
+            reason="unlock other pr",
+            operator_authorized=True,
+            target_pr_number=692,
+        )
+        self.assertFalse(r["authorized"])
+        self.assertTrue(any("different PR" in x for x in r["reasons"]))
+
     def test_authorize_review_correction_requires_operator_auth(self):
         from mcp_server import _load_review_decision_lock, _save_review_decision_lock
         lock = _load_review_decision_lock() or {}
@@ -2615,6 +2631,7 @@ class TestSubmitPrReview(unittest.TestCase):
             prior_review_state="approve",
             reason="typo",
             operator_authorized=False,
+            target_pr_number=8,
         )
         self.assertFalse(r["authorized"])
         self.assertTrue(any("operator authorization" in x for x in r["reasons"]))
@@ -2672,6 +2689,7 @@ class TestSubmitPrReview(unittest.TestCase):
                     prior_review_state="approve",
                     reason="operator approved correcting mistaken approve",
                     operator_authorized=True,
+                    target_pr_number=8,
                 )
                 _mark_request_changes_ready(remote="prgs")
                 second = gitea_submit_pr_review(

tests/test_merger_lease_acquire.py

@@ -0,0 +1,466 @@
+"""Merger lease acquisition + capability resolution tests (#718, #723)."""
+
+from __future__ import annotations
+
+import os
+import sys
+import unittest
+from datetime import datetime, timezone
+from unittest.mock import MagicMock, patch
+
+sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
+
+import gitea_mcp_server as mcp_server
+import reviewer_pr_lease as leases
+import task_capability_map as tcm
+
+
+HEAD = "a" * 40
+LIVE_HEAD = HEAD
+
+
+class TestMergerLeaseAcquireTool(unittest.TestCase):
+    def setUp(self):
+        mcp_server._preflight_whoami_called = True
+        mcp_server._preflight_capability_called = True
+        mcp_server._preflight_resolved_role = "merger"
+        mcp_server._preflight_resolved_task = "acquire_merger_pr_lease"
+        leases.clear_session_lease()
+
+    def tearDown(self):
+        leases.clear_session_lease()
+
+    def _merger_profile(self, **extra):
+        p = {
+            "username": "merger-user",
+            "profile_name": "prgs-merger",
+            "role": "merger",
+            "allowed_operations": [
+                "gitea.read",
+                "gitea.pr.comment",
+                "gitea.pr.merge",
+            ],
+            "forbidden_operations": [
+                "gitea.pr.approve",
+                "gitea.pr.review",
+                "gitea.pr.request_changes",
+            ],
+        }
+        p.update(extra)
+        return p
+
+    @patch("gitea_mcp_server.api_request")
+    @patch("gitea_mcp_server._auth", return_value="token pass")
+    @patch("gitea_mcp_server._resolve", return_value=("gitea.prgs.cc", "Scaled-Tech-Consulting", "Gitea-Tools"))
+    @patch("gitea_mcp_server.get_profile")
+    @patch("gitea_mcp_server._fetch_pr_comments", return_value=[])
+    @patch("gitea_mcp_server._verify_role_mutation_workspace")
+    def test_acquire_merger_lease_success(
+        self, _verify, _comments, mock_profile, _resolve, _auth, mock_api
+    ):
+        mock_profile.return_value = self._merger_profile()
+        mock_api.side_effect = [
+            {"state": "open", "head": {"sha": LIVE_HEAD}},
+            {"id": 456},
+        ]
+        with patch.dict(os.environ, {"GITEA_MCP_PROFILE_NAME": "prgs-merger"}):
+            res = mcp_server.gitea_acquire_merger_pr_lease(
+                pr_number=718,
+                worktree="branches/issue-718",
+                session_id="merger-session",
+                candidate_head=HEAD,
+                remote="prgs",
+                org="Scaled-Tech-Consulting",
+                repo="Gitea-Tools",
+            )
+        self.assertTrue(res["success"], res)
+        self.assertTrue(res["acquired"])
+        self.assertEqual(res["comment_id"], 456)
+        self.assertEqual(res["session_id"], "merger-session")
+        self.assertEqual(res.get("repo"), "Scaled-Tech-Consulting/Gitea-Tools")
+        body = mock_api.call_args_list[-1][0][3]["body"]
+        self.assertIn("reviewer_identity: merger-user", body)
+        self.assertIn("profile: prgs-merger", body)
+        session_lease = leases._SESSION_LEASE
+        self.assertIsNotNone(session_lease)
+        provenance = session_lease.get("lease_provenance") or {}
+        self.assertEqual(provenance.get("source"), "gitea_acquire_merger_pr_lease")
+
+    @patch("gitea_mcp_server.api_request")
+    @patch("gitea_mcp_server._auth", return_value="token pass")
+    @patch("gitea_mcp_server._resolve", return_value=("gitea.prgs.cc", "Scaled-Tech-Consulting", "Gitea-Tools"))
+    @patch("gitea_mcp_server.get_profile")
+    @patch("gitea_mcp_server._fetch_pr_comments")
+    @patch("gitea_mcp_server._verify_role_mutation_workspace")
+    def test_acquire_merger_lease_fails_if_active_exists(
+        self, _verify, _comments, mock_profile, _resolve, _auth, mock_api
+    ):
+        mock_profile.return_value = self._merger_profile()
+        active_body = leases.format_lease_body(
+            repo="Scaled-Tech-Consulting/Gitea-Tools",
+            pr_number=718,
+            issue_number=None,
+            reviewer_identity="other-user",
+            profile="prgs-reviewer",
+            session_id="other-session",
+            worktree="branches/review-1",
+            phase="claimed",
+            candidate_head=HEAD,
+            target_branch="master",
+            target_branch_sha="b" * 40,
+            last_activity=datetime.now(timezone.utc),
+        )
+        _comments.return_value = [
+            {"id": 1, "body": active_body, "user": {"login": "other-user"}}
+        ]
+        mock_api.return_value = {"state": "open", "head": {"sha": LIVE_HEAD}}
+        with patch.dict(os.environ, {"GITEA_MCP_PROFILE_NAME": "prgs-merger"}):
+            res = mcp_server.gitea_acquire_merger_pr_lease(
+                pr_number=718,
+                worktree="branches/issue-718",
+                session_id="merger-session",
+                candidate_head=HEAD,
+            )
+        self.assertFalse(res["success"])
+        self.assertFalse(res["acquired"])
+        self.assertIn("already has active reviewer lease", " ".join(res["reasons"]))
+        self.assertEqual(
+            [c for c in mock_api.call_args_list if c[0][0] == "POST"], []
+        )
+        self.assertIsNone(leases._SESSION_LEASE)
+
+    @patch("gitea_mcp_server.get_profile")
+    @patch("gitea_mcp_server._verify_role_mutation_workspace")
+    def test_acquire_merger_lease_fails_workspace_binding(self, _verify, mock_profile):
+        mock_profile.return_value = self._merger_profile()
+        _verify.side_effect = RuntimeError("Branches-only mutation guard")
+        with patch.dict(os.environ, {"GITEA_MCP_PROFILE_NAME": "prgs-merger"}):
+            res = mcp_server.gitea_acquire_merger_pr_lease(
+                pr_number=718,
+                worktree="branches/issue-718",
+                session_id="merger-session",
+                candidate_head=HEAD,
+            )
+        self.assertFalse(res["success"])
+        self.assertIn("Branches-only mutation guard", res["reasons"][0])
+        self.assertIsNone(leases._SESSION_LEASE)
+
+    @patch("gitea_mcp_server.api_request")
+    @patch("gitea_mcp_server._auth", return_value="token pass")
+    @patch("gitea_mcp_server._resolve", return_value=("gitea.prgs.cc", "Scaled-Tech-Consulting", "Gitea-Tools"))
+    @patch("gitea_mcp_server.get_profile")
+    @patch("gitea_mcp_server._fetch_pr_comments", return_value=[])
+    @patch("gitea_mcp_server._verify_role_mutation_workspace")
+    def test_acquire_merger_forwards_explicit_org_repo_to_workspace_verify(
+        self, mock_verify, _comments, mock_profile, _resolve, _auth, mock_api
+    ):
+        """Explicit org/repo must reach anti-stomp (prgs bare default is Timesheet)."""
+        mock_profile.return_value = self._merger_profile()
+        mock_api.side_effect = [
+            {"state": "open", "head": {"sha": LIVE_HEAD}},
+            {"id": 789},
+        ]
+        with patch.dict(os.environ, {"GITEA_MCP_PROFILE_NAME": "prgs-merger"}):
+            res = mcp_server.gitea_acquire_merger_pr_lease(
+                pr_number=718,
+                worktree="branches/issue-718",
+                session_id="merger-session",
+                candidate_head=HEAD,
+                remote="prgs",
+                org="Scaled-Tech-Consulting",
+                repo="Gitea-Tools",
+            )
+        self.assertTrue(res["success"], res)
+        kwargs = mock_verify.call_args.kwargs
+        self.assertEqual(kwargs.get("org"), "Scaled-Tech-Consulting")
+        self.assertEqual(kwargs.get("repo"), "Gitea-Tools")
+        self.assertEqual(kwargs.get("task"), "acquire_merger_pr_lease")
+
+    @patch("gitea_mcp_server.get_profile")
+    def test_acquire_merger_requires_candidate_head(self, mock_profile):
+        mock_profile.return_value = self._merger_profile()
+        res = mcp_server.gitea_acquire_merger_pr_lease(
+            pr_number=718,
+            worktree="branches/issue-718",
+            candidate_head=None,
+        )
+        self.assertFalse(res["success"])
+        self.assertTrue(any("candidate_head is required" in r for r in res["reasons"]))
+
+    @patch("gitea_mcp_server.api_request")
+    @patch("gitea_mcp_server._auth", return_value="token pass")
+    @patch("gitea_mcp_server._resolve", return_value=("gitea.prgs.cc", "Scaled-Tech-Consulting", "Gitea-Tools"))
+    @patch("gitea_mcp_server.get_profile")
+    @patch("gitea_mcp_server._fetch_pr_comments", return_value=[])
+    @patch("gitea_mcp_server._verify_role_mutation_workspace")
+    def test_acquire_merger_head_mismatch(
+        self, _verify, _comments, mock_profile, _resolve, _auth, mock_api
+    ):
+        mock_profile.return_value = self._merger_profile()
+        mock_api.return_value = {"state": "open", "head": {"sha": "b" * 40}}
+        res = mcp_server.gitea_acquire_merger_pr_lease(
+            pr_number=718,
+            worktree="branches/issue-718",
+            candidate_head=HEAD,
+        )
+        self.assertFalse(res["success"])
+        self.assertTrue(any("does not match live PR head" in r for r in res["reasons"]))
+        self.assertIsNone(leases._SESSION_LEASE)
+
+    @patch("gitea_mcp_server.api_request")
+    @patch("gitea_mcp_server._auth", return_value="token pass")
+    @patch("gitea_mcp_server._resolve", return_value=("gitea.prgs.cc", "Scaled-Tech-Consulting", "Gitea-Tools"))
+    @patch("gitea_mcp_server.get_profile")
+    @patch("gitea_mcp_server._fetch_pr_comments", return_value=[])
+    @patch("gitea_mcp_server._verify_role_mutation_workspace")
+    def test_acquire_merger_fails_closed_when_live_head_unresolvable(
+        self, _verify, _comments, mock_profile, _resolve, _auth, mock_api
+    ):
+        """Empty/missing live head must not silently proceed past exact-head gate."""
+        mock_profile.return_value = self._merger_profile()
+        # PR payload present but head.sha missing / empty / non-dict.
+        mock_api.return_value = {"state": "open", "head": {}}
+        res = mcp_server.gitea_acquire_merger_pr_lease(
+            pr_number=718,
+            worktree="branches/issue-718",
+            session_id="merger-session",
+            candidate_head=HEAD,
+        )
+        self.assertFalse(res["success"], res)
+        self.assertFalse(res.get("acquired"), res)
+        self.assertTrue(
+            any("live PR head could not be resolved" in r for r in res["reasons"]),
+            res,
+        )
+        self.assertIsNone(res.get("live_head"))
+        self.assertEqual(res.get("candidate_head"), HEAD)
+        self.assertEqual(
+            [c for c in mock_api.call_args_list if c[0][0] == "POST"], []
+        )
+        self.assertIsNone(leases._SESSION_LEASE)
+
+        # Completely empty GET /pulls body also fails closed.
+        mock_api.return_value = {}
+        res2 = mcp_server.gitea_acquire_merger_pr_lease(
+            pr_number=718,
+            worktree="branches/issue-718",
+            candidate_head=HEAD,
+        )
+        self.assertFalse(res2["success"], res2)
+        self.assertTrue(
+            any("live PR head could not be resolved" in r for r in res2["reasons"]),
+            res2,
+        )
+        self.assertIsNone(leases._SESSION_LEASE)
+
+    @patch("gitea_mcp_server._profile_operation_gate")
+    def test_author_denied_merge_permission(self, mock_gate):
+        """Author profile lacks gitea.pr.merge → fail closed before mutation."""
+        def gate(op):
+            if op == "gitea.pr.merge":
+                return [f"profile lacks {op}"]
+            return None
+        mock_gate.side_effect = gate
+        res = mcp_server.gitea_acquire_merger_pr_lease(
+            pr_number=718,
+            worktree="branches/issue-718",
+            candidate_head=HEAD,
+        )
+        self.assertFalse(res["success"])
+        self.assertTrue(any("gitea.pr.merge" in r for r in res["reasons"]))
+
+    @patch("gitea_mcp_server._profile_operation_gate")
+    def test_reviewer_denied_merge_permission(self, mock_gate):
+        def gate(op):
+            if op == "gitea.pr.merge":
+                return [f"profile lacks {op}"]
+            return None
+        mock_gate.side_effect = gate
+        res = mcp_server.gitea_acquire_merger_pr_lease(
+            pr_number=718,
+            worktree="branches/issue-718",
+            candidate_head=HEAD,
+        )
+        self.assertFalse(res["success"])
+        self.assertTrue(any("gitea.pr.merge" in r for r in res["reasons"]))
+
+
+class TestCapabilityMapLeaseTasks(unittest.TestCase):
+    def test_merger_acquire_map_entries(self):
+        for task in ("acquire_merger_pr_lease", "gitea_acquire_merger_pr_lease"):
+            self.assertEqual(tcm.required_permission(task), "gitea.pr.comment")
+            self.assertEqual(tcm.required_role(task), "merger")
+
+    def test_reviewer_acquire_map_entries(self):
+        for task in ("acquire_reviewer_pr_lease", "gitea_acquire_reviewer_pr_lease"):
+            self.assertEqual(tcm.required_permission(task), "gitea.pr.comment")
+            self.assertEqual(tcm.required_role(task), "reviewer")
+
+    def test_adopt_merger_aliases(self):
+        for task in ("adopt_merger_pr_lease", "gitea_adopt_merger_pr_lease"):
+            self.assertEqual(tcm.required_role(task), "merger")
+
+
+class TestResolveTaskCapabilityLeaseAndUnknown(unittest.TestCase):
+    def setUp(self):
+        mcp_server._process_boot_head_sha = None
+        if hasattr(mcp_server, "capability_stop_terminal"):
+            mcp_server.capability_stop_terminal.clear()
+
+    @patch.object(mcp_server, "record_mutation_authority", return_value=None)
+    @patch.object(mcp_server, "init_review_decision_lock", return_value=None)
+    @patch.object(mcp_server, "record_preflight_check", return_value=None)
+    @patch.object(mcp_server, "_ensure_matching_profile", return_value=None)
+    @patch.object(mcp_server, "_authenticated_username", return_value="sysadmin")
+    @patch.object(mcp_server, "get_profile")
+    @patch.object(mcp_server.gitea_config, "load_config")
+    def test_resolve_acquire_reviewer_authorized(
+        self, mock_cfg, mock_profile, *_mocks
+    ):
+        mock_profile.return_value = {
+            "profile_name": "prgs-reviewer",
+            "role": "reviewer",
+            "allowed_operations": ["gitea.pr.comment", "gitea.read", "gitea.pr.review"],
+            "forbidden_operations": [],
+        }
+        mock_cfg.return_value = {
+            "profiles": {
+                "prgs-reviewer": {
+                    "role": "reviewer",
+                    "allowed_operations": [
+                        "gitea.pr.comment",
+                        "gitea.read",
+                        "gitea.pr.review",
+                    ],
+                    "forbidden_operations": [],
+                }
+            }
+        }
+        with patch.dict(os.environ, {"GITEA_MCP_PROFILE": "prgs-reviewer"}, clear=False):
+            for task in ("acquire_reviewer_pr_lease", "gitea_acquire_reviewer_pr_lease"):
+                res = mcp_server.gitea_resolve_task_capability(task=task, remote="prgs")
+                self.assertEqual(res["required_role_kind"], "reviewer", res)
+                self.assertEqual(
+                    res["required_operation_permission"], "gitea.pr.comment", res
+                )
+                self.assertTrue(res.get("allowed_in_current_session"), res)
+
+    @patch.object(mcp_server, "record_mutation_authority", return_value=None)
+    @patch.object(mcp_server, "init_review_decision_lock", return_value=None)
+    @patch.object(mcp_server, "record_preflight_check", return_value=None)
+    @patch.object(mcp_server, "_ensure_matching_profile", return_value=None)
+    @patch.object(mcp_server, "_authenticated_username", return_value="jcwalker3")
+    @patch.object(mcp_server, "get_profile")
+    @patch.object(mcp_server.gitea_config, "load_config")
+    def test_resolve_acquire_reviewer_author_denied(
+        self, mock_cfg, mock_profile, *_mocks
+    ):
+        mock_profile.return_value = {
+            "profile_name": "prgs-author",
+            "role": "author",
+            "allowed_operations": ["gitea.pr.comment", "gitea.branch.push", "gitea.read"],
+            "forbidden_operations": [],
+        }
+        mock_cfg.return_value = {
+            "profiles": {
+                "prgs-author": {
+                    "role": "author",
+                    "allowed_operations": [
+                        "gitea.pr.comment",
+                        "gitea.branch.push",
+                        "gitea.read",
+                    ],
+                    "forbidden_operations": [],
+                },
+                "prgs-reviewer": {
+                    "role": "reviewer",
+                    "allowed_operations": ["gitea.pr.comment", "gitea.pr.review"],
+                    "forbidden_operations": [],
+                },
+            }
+        }
+        with patch.dict(os.environ, {"GITEA_MCP_PROFILE": "prgs-author"}, clear=False):
+            res = mcp_server.gitea_resolve_task_capability(
+                task="acquire_reviewer_pr_lease", remote="prgs"
+            )
+        self.assertFalse(res.get("allowed_in_current_session"), res)
+        self.assertEqual(res["required_role_kind"], "reviewer")
+        guidance = " ".join(res.get("task_role_guidance") or [])
+        self.assertTrue(
+            "cannot perform reviewer" in guidance.lower()
+            or "reviewer" in guidance.lower()
+            or res.get("stop_required"),
+            res,
+        )
+
+    @patch.object(mcp_server, "record_mutation_authority", return_value=None)
+    @patch.object(mcp_server, "init_review_decision_lock", return_value=None)
+    @patch.object(mcp_server, "record_preflight_check", return_value=None)
+    @patch.object(mcp_server, "_ensure_matching_profile", return_value=None)
+    @patch.object(mcp_server, "_authenticated_username", return_value="sysadmin")
+    @patch.object(mcp_server, "get_profile")
+    def test_resolve_unknown_task_structured_no_raise(self, mock_profile, *_mocks):
+        mock_profile.return_value = {
+            "profile_name": "prgs-reviewer",
+            "role": "reviewer",
+            "allowed_operations": ["gitea.pr.comment"],
+            "forbidden_operations": [],
+        }
+        # Must not raise ValueError into the tool boundary.
+        res = mcp_server.gitea_resolve_task_capability(
+            task="some_made_up_task", remote="prgs"
+        )
+        self.assertIsInstance(res, dict)
+        self.assertEqual(res.get("reason_code"), "unknown_task", res)
+        self.assertFalse(res.get("allowed_in_current_session"))
+        self.assertTrue(res.get("stop_required"))
+        self.assertIs(res.get("mutation_performed"), False)
+        self.assertNotIn("token", str(res).lower())
+        self.assertNotIn("password", str(res).lower())
+        guidance = " ".join(res.get("task_role_guidance") or [])
+        self.assertIn("Unknown task/action", guidance)
+
+    @patch.object(mcp_server, "record_mutation_authority", return_value=None)
+    @patch.object(mcp_server, "init_review_decision_lock", return_value=None)
+    @patch.object(mcp_server, "record_preflight_check", return_value=None)
+    @patch.object(mcp_server, "_ensure_matching_profile", return_value=None)
+    @patch.object(mcp_server, "_authenticated_username", return_value="sysadmin")
+    @patch.object(mcp_server, "get_profile")
+    @patch.object(mcp_server.gitea_config, "load_config")
+    def test_resolve_merger_acquire_aliases(
+        self, mock_cfg, mock_profile, *_mocks
+    ):
+        mock_profile.return_value = {
+            "profile_name": "prgs-merger",
+            "role": "merger",
+            "allowed_operations": [
+                "gitea.read",
+                "gitea.pr.comment",
+                "gitea.pr.merge",
+            ],
+            "forbidden_operations": [],
+        }
+        mock_cfg.return_value = {
+            "profiles": {
+                "prgs-merger": {
+                    "role": "merger",
+                    "allowed_operations": [
+                        "gitea.read",
+                        "gitea.pr.comment",
+                        "gitea.pr.merge",
+                    ],
+                    "forbidden_operations": [],
+                }
+            }
+        }
+        with patch.dict(os.environ, {"GITEA_MCP_PROFILE": "prgs-merger"}, clear=False):
+            for task in ("acquire_merger_pr_lease", "gitea_acquire_merger_pr_lease"):
+                res = mcp_server.gitea_resolve_task_capability(task=task, remote="prgs")
+                self.assertEqual(res["required_role_kind"], "merger", res)
+                self.assertEqual(
+                    res["required_operation_permission"], "gitea.pr.comment", res
+                )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_pr_ownership_issue_pr_mismatch.py

@@ -0,0 +1,196 @@
+"""#727 / PR #728: author ownership when tracking issue number != PR number.
+
+Regression for REQUEST_CHANGES: gitea_update_pr_branch_by_merge must not require
+issue_number == pr_number. Ownership is proven via linked issue + lock/branch
+context and fails closed for unrelated issues.
+"""
+
+from __future__ import annotations
+
+import os
+import tempfile
+import unittest
+from datetime import datetime, timedelta, timezone
+from unittest.mock import patch
+
+import issue_lock_store
+import gitea_mcp_server as mcp
+
+
+def _live_lock(
+    *,
+    issue_number: int,
+    branch_name: str = "feat/issue-727-pr-sync-status",
+    worktree_path: str = "/tmp/branches/issue-727-pr-sync-status",
+    remote: str = "prgs",
+    org: str = "Scaled-Tech-Consulting",
+    repo: str = "Gitea-Tools",
+) -> dict:
+    now = datetime.now(timezone.utc)
+    return {
+        "issue_number": issue_number,
+        "branch_name": branch_name,
+        "worktree_path": worktree_path,
+        "remote": remote,
+        "org": org,
+        "repo": repo,
+        "operation_type": issue_lock_store.AUTHOR_ISSUE_WORK_LEASE,
+        "acquired_at": now.isoformat(),
+        "expires_at": (now + timedelta(hours=2)).isoformat(),
+        "owner_pid": os.getpid(),
+        "status": "active",
+    }
+
+
+class TestAuthorOwnershipIssuePrMismatch(unittest.TestCase):
+    def setUp(self):
+        self._tmp = tempfile.TemporaryDirectory(prefix="gitea-issue-locks-")
+        self.lock_dir = self._tmp.name
+        os.environ["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir
+
+    def tearDown(self):
+        # Clear session pointer for this process.
+        try:
+            ptr = issue_lock_store.session_pointer_path(self.lock_dir)
+            if os.path.isfile(ptr):
+                os.unlink(ptr)
+        except Exception:
+            pass
+        os.environ.pop("GITEA_ISSUE_LOCK_DIR", None)
+        self._tmp.cleanup()
+
+    def test_issue_727_pr_728_session_lock_accepted(self):
+        """Tracking issue #727 lock is valid ownership for PR #728."""
+        lock = _live_lock(issue_number=727)
+        issue_lock_store.bind_session_lock(lock)
+        result = mcp._prove_author_ownership_for_pr(
+            pr_number=728,
+            pr_title="feat: pr sync",
+            pr_body="Closes #727\n\nNative PR sync lifecycle.",
+            source_branch="feat/issue-727-pr-sync-status",
+            remote="prgs",
+            host=None,
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+            worktree_path=lock["worktree_path"],
+        )
+        self.assertTrue(result["proven"], result)
+        self.assertTrue(result["has_author_lock"])
+        self.assertEqual(result["matched_issue"], 727)
+        self.assertEqual(result["matched_via"], "session_lock")
+        self.assertIn(727, result["linked_issues"])
+
+    def test_issue_727_pr_728_durable_lock_accepted(self):
+        lock = _live_lock(issue_number=727)
+        path = issue_lock_store.lock_file_path(
+            remote="prgs",
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+            issue_number=727,
+            lock_dir=self.lock_dir,
+        )
+        issue_lock_store.save_lock_file(path, lock)
+        result = mcp._prove_author_ownership_for_pr(
+            pr_number=728,
+            pr_title="feat: pr sync",
+            pr_body="Fixes #727",
+            source_branch="feat/issue-727-pr-sync-status",
+            remote="prgs",
+            host=None,
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+        )
+        self.assertTrue(result["proven"], result)
+        self.assertEqual(result["matched_issue"], 727)
+        self.assertEqual(result["matched_via"], "durable_lock")
+
+    def test_legacy_same_number_still_accepted(self):
+        lock = _live_lock(
+            issue_number=100,
+            branch_name="fix/issue-100",
+            worktree_path="/tmp/branches/issue-100",
+        )
+        issue_lock_store.bind_session_lock(lock)
+        result = mcp._prove_author_ownership_for_pr(
+            pr_number=100,
+            pr_title="fix something",
+            pr_body="Closes #100",
+            source_branch="fix/issue-100",
+            remote="prgs",
+            host=None,
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+            worktree_path=lock["worktree_path"],
+        )
+        self.assertTrue(result["proven"], result)
+        self.assertEqual(result["matched_issue"], 100)
+
+    def test_unrelated_issue_lock_rejected(self):
+        """Lock for issue #999 must not authorize PR #728 linked only to #727."""
+        lock = _live_lock(
+            issue_number=999,
+            branch_name="feat/issue-999",
+            worktree_path="/tmp/branches/issue-999",
+        )
+        issue_lock_store.bind_session_lock(lock)
+        path = issue_lock_store.lock_file_path(
+            remote="prgs",
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+            issue_number=999,
+            lock_dir=self.lock_dir,
+        )
+        issue_lock_store.save_lock_file(path, lock)
+        result = mcp._prove_author_ownership_for_pr(
+            pr_number=728,
+            pr_title="feat: pr sync",
+            pr_body="Closes #727",
+            source_branch="feat/issue-727-pr-sync-status",
+            remote="prgs",
+            host=None,
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+            worktree_path=lock["worktree_path"],
+        )
+        self.assertFalse(result["proven"], result)
+        self.assertFalse(result["has_author_lock"])
+        self.assertIsNone(result["matched_issue"])
+        self.assertTrue(result["reasons"])
+
+    def test_branch_mismatch_on_session_lock_fail_closed(self):
+        lock = _live_lock(
+            issue_number=727,
+            branch_name="feat/other-branch",
+        )
+        issue_lock_store.bind_session_lock(lock)
+        result = mcp._prove_author_ownership_for_pr(
+            pr_number=728,
+            pr_title="feat: pr sync",
+            pr_body="Closes #727",
+            source_branch="feat/issue-727-pr-sync-status",
+            remote="prgs",
+            host=None,
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+            worktree_path=lock["worktree_path"],
+        )
+        self.assertFalse(result["proven"], result)
+        self.assertTrue(any("branch" in r for r in result["reasons"]))
+
+    def test_no_lock_fail_closed(self):
+        result = mcp._prove_author_ownership_for_pr(
+            pr_number=728,
+            pr_title="feat: pr sync",
+            pr_body="Closes #727",
+            source_branch="feat/issue-727-pr-sync-status",
+            remote="prgs",
+            host=None,
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+        )
+        self.assertFalse(result["proven"], result)
+        self.assertTrue(any("no live author issue lock" in r for r in result["reasons"]))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_pr_sync_status.py

@@ -0,0 +1,338 @@
+"""Hermetic tests for PR synchronization / conflict-remediation lifecycle."""
+
+from __future__ import annotations
+
+import unittest
+
+from pr_sync_status import (
+    ACTION_AUTHOR_CONFLICT_REMEDIATION,
+    ACTION_BLOCKED,
+    ACTION_FRESH_REVIEW_REQUIRED,
+    ACTION_MERGE_NOW,
+    ACTION_UPDATE_BRANCH_BY_MERGE,
+    assess_post_update_head_transition,
+    assess_pr_sync_status,
+    assess_sequential_queue_step,
+    assess_update_pr_branch_preflight,
+    lease_usable_at_head,
+    merge_approval_usable_at_head,
+)
+
+
+def _sha(prefix: str) -> str:
+    return (prefix + "0" * 40)[:40]
+
+
+PR_HEAD = _sha("aaaaaaaa")
+BASE_HEAD = _sha("bbbbbbbb")
+NEW_HEAD = _sha("cccccccc")
+OLD_HEAD = _sha("dddddddd")
+
+
+def _base_kwargs(**overrides):
+    data = {
+        "host": "gitea.prgs.cc",
+        "org": "Scaled-Tech-Consulting",
+        "repo": "Gitea-Tools",
+        "pr_number": 100,
+        "pr_state": "open",
+        "source_branch": "fix/issue-100",
+        "pr_head_sha": PR_HEAD,
+        "base_head_sha": BASE_HEAD,
+        "commits_behind": 0,
+        "mergeable": True,
+        "has_conflicts": False,
+        "branch_protection_requires_current_base": False,
+        "approval_at_current_head": True,
+        "checks_status": "success",
+        "active_author_lock": True,
+        "active_reviewer_lease": False,
+        "active_merger_lease": False,
+    }
+    data.update(overrides)
+    return data
+
+
+class TestAssessPrSyncStatus(unittest.TestCase):
+    def test_approved_current_mergeable_merge_now(self):
+        result = assess_pr_sync_status(**_base_kwargs())
+        self.assertEqual(result["recommended_next_action"], ACTION_MERGE_NOW)
+        self.assertTrue(result["approval_valid_for_merge"])
+        self.assertFalse(result["stale_approval"])
+        self.assertEqual(result["pr_head_sha"], PR_HEAD)
+        self.assertEqual(result["base_head_sha"], BASE_HEAD)
+
+    def test_approved_outdated_update_not_required_merge_now(self):
+        result = assess_pr_sync_status(
+            **_base_kwargs(
+                commits_behind=3,
+                branch_protection_requires_current_base=False,
+            )
+        )
+        self.assertEqual(result["recommended_next_action"], ACTION_MERGE_NOW)
+        self.assertTrue(result["approval_valid_for_merge"])
+        self.assertTrue(any("does not require" in r for r in result["reasons"]))
+
+    def test_outdated_update_required_no_conflict(self):
+        result = assess_pr_sync_status(
+            **_base_kwargs(
+                commits_behind=2,
+                branch_protection_requires_current_base=True,
+                mergeable=True,
+                has_conflicts=False,
+            )
+        )
+        self.assertEqual(
+            result["recommended_next_action"], ACTION_UPDATE_BRANCH_BY_MERGE
+        )
+        self.assertFalse(result["approval_valid_for_merge"])
+        self.assertTrue(any("update_branch_by_merge" in r for r in result["reasons"]))
+
+    def test_outdated_has_conflicts_author_remediation(self):
+        result = assess_pr_sync_status(
+            **_base_kwargs(
+                commits_behind=5,
+                branch_protection_requires_current_base=True,
+                mergeable=False,
+                has_conflicts=True,
+            )
+        )
+        self.assertEqual(
+            result["recommended_next_action"], ACTION_AUTHOR_CONFLICT_REMEDIATION
+        )
+        self.assertFalse(result["approval_valid_for_merge"])
+
+    def test_stale_approval_fresh_review_required(self):
+        result = assess_pr_sync_status(
+            **_base_kwargs(
+                approval_at_current_head=False,
+                commits_behind=0,
+            )
+        )
+        self.assertEqual(
+            result["recommended_next_action"], ACTION_FRESH_REVIEW_REQUIRED
+        )
+        self.assertTrue(result["stale_approval"])
+        self.assertFalse(result["approval_valid_for_merge"])
+
+    def test_stale_prepared_verdict_cannot_cross_heads(self):
+        result = assess_pr_sync_status(
+            **_base_kwargs(
+                approval_at_current_head=True,
+                prepared_verdict_head_sha=OLD_HEAD,
+            )
+        )
+        self.assertEqual(
+            result["recommended_next_action"], ACTION_FRESH_REVIEW_REQUIRED
+        )
+        self.assertTrue(result["stale_prepared_verdict"])
+        self.assertFalse(result["approval_valid_for_merge"])
+
+    def test_missing_heads_fail_closed(self):
+        result = assess_pr_sync_status(
+            **_base_kwargs(pr_head_sha="short", base_head_sha=None)
+        )
+        self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED)
+        self.assertTrue(any("PR head" in r for r in result["reasons"]))
+
+    def test_closed_pr_blocked(self):
+        result = assess_pr_sync_status(**_base_kwargs(pr_state="closed"))
+        self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED)
+
+    def test_failed_checks_block_merge_now(self):
+        result = assess_pr_sync_status(**_base_kwargs(checks_status="failure"))
+        self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED)
+
+    def test_stale_approval_with_update_required_routes_update(self):
+        result = assess_pr_sync_status(
+            **_base_kwargs(
+                approval_at_current_head=False,
+                commits_behind=1,
+                branch_protection_requires_current_base=True,
+                mergeable=True,
+                has_conflicts=False,
+            )
+        )
+        self.assertEqual(
+            result["recommended_next_action"], ACTION_UPDATE_BRANCH_BY_MERGE
+        )
+
+
+class TestUpdatePrBranchPreflight(unittest.TestCase):
+    def _ok_kwargs(self, **overrides):
+        data = {
+            "role_kind": "author",
+            "expected_pr_head_sha": PR_HEAD,
+            "live_pr_head_sha": PR_HEAD,
+            "expected_base_head_sha": BASE_HEAD,
+            "live_base_head_sha": BASE_HEAD,
+            "has_conflicts": False,
+            "mergeable": True,
+            "style": "merge",
+            "has_author_lock": True,
+            "worktree_path": "/Users/x/Development/Gitea-Tools/branches/issue-100",
+            "worktree_owns_source_branch": True,
+            "control_checkout_clean": True,
+            "master_parity_ok": True,
+            "force_push": False,
+            "rebase": False,
+        }
+        data.update(overrides)
+        return data
+
+    def test_author_allowed_when_preflight_clean(self):
+        result = assess_update_pr_branch_preflight(**self._ok_kwargs())
+        self.assertTrue(result["mutation_allowed"])
+        self.assertFalse(result["head_race"])
+        self.assertFalse(result["base_race"])
+
+    def test_head_race_fail_closed(self):
+        result = assess_update_pr_branch_preflight(
+            **self._ok_kwargs(live_pr_head_sha=NEW_HEAD)
+        )
+        self.assertFalse(result["mutation_allowed"])
+        self.assertTrue(result["head_race"])
+        self.assertTrue(any("PR head race" in r for r in result["reasons"]))
+
+    def test_base_race_fail_closed(self):
+        result = assess_update_pr_branch_preflight(
+            **self._ok_kwargs(live_base_head_sha=NEW_HEAD)
+        )
+        self.assertFalse(result["mutation_allowed"])
+        self.assertTrue(result["base_race"])
+        self.assertTrue(any("base head race" in r for r in result["reasons"]))
+
+    def test_conflicts_structured_handoff_no_mutation(self):
+        result = assess_update_pr_branch_preflight(
+            **self._ok_kwargs(has_conflicts=True, mergeable=False)
+        )
+        self.assertFalse(result["mutation_allowed"])
+        self.assertTrue(result["author_remediation_handoff"])
+        self.assertTrue(any("conflicts" in r.lower() for r in result["reasons"]))
+
+    def test_reviewer_denied(self):
+        result = assess_update_pr_branch_preflight(
+            **self._ok_kwargs(role_kind="reviewer")
+        )
+        self.assertFalse(result["mutation_allowed"])
+        self.assertTrue(any("author-only" in r for r in result["reasons"]))
+
+    def test_merger_denied(self):
+        result = assess_update_pr_branch_preflight(
+            **self._ok_kwargs(role_kind="merger")
+        )
+        self.assertFalse(result["mutation_allowed"])
+        self.assertTrue(any("author-only" in r for r in result["reasons"]))
+
+    def test_no_force_push(self):
+        result = assess_update_pr_branch_preflight(
+            **self._ok_kwargs(force_push=True)
+        )
+        self.assertFalse(result["mutation_allowed"])
+        self.assertTrue(any("force-push" in r for r in result["reasons"]))
+
+    def test_no_rebase(self):
+        result = assess_update_pr_branch_preflight(
+            **self._ok_kwargs(style="rebase", rebase=True)
+        )
+        self.assertFalse(result["mutation_allowed"])
+        self.assertTrue(any("rebase" in r for r in result["reasons"]))
+
+    def test_missing_lock_fail_closed(self):
+        result = assess_update_pr_branch_preflight(
+            **self._ok_kwargs(has_author_lock=False)
+        )
+        self.assertFalse(result["mutation_allowed"])
+
+    def test_worktree_not_under_branches(self):
+        result = assess_update_pr_branch_preflight(
+            **self._ok_kwargs(worktree_path="/tmp/not-a-branches-path")
+        )
+        self.assertFalse(result["mutation_allowed"])
+        self.assertTrue(any("branches/" in r for r in result["reasons"]))
+
+
+class TestPostUpdateHeadTransition(unittest.TestCase):
+    def test_update_invalidates_approval_and_requires_fresh_review(self):
+        result = assess_post_update_head_transition(
+            former_pr_head_sha=PR_HEAD,
+            new_pr_head_sha=NEW_HEAD,
+            former_approval_head_sha=PR_HEAD,
+            former_reviewer_lease_head_sha=PR_HEAD,
+            former_merger_lease_head_sha=PR_HEAD,
+            prepared_verdict_head_sha=PR_HEAD,
+        )
+        self.assertTrue(result["head_changed"])
+        self.assertTrue(result["approval_invalidated"])
+        self.assertTrue(result["reviewer_lease_superseded"])
+        self.assertTrue(result["merger_lease_superseded"])
+        self.assertTrue(result["prepared_verdict_invalidated"])
+        self.assertEqual(
+            result["recommended_next_action"], ACTION_FRESH_REVIEW_REQUIRED
+        )
+
+    def test_same_head_unexpected(self):
+        result = assess_post_update_head_transition(
+            former_pr_head_sha=PR_HEAD,
+            new_pr_head_sha=PR_HEAD,
+        )
+        self.assertFalse(result["head_changed"])
+        self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED)
+
+
+class TestStaleArtifacts(unittest.TestCase):
+    def test_stale_approval_cannot_authorize_merge(self):
+        result = merge_approval_usable_at_head(
+            current_head_sha=NEW_HEAD,
+            approved_head_sha=PR_HEAD,
+        )
+        self.assertFalse(result["approval_usable"])
+
+    def test_fresh_approval_usable(self):
+        result = merge_approval_usable_at_head(
+            current_head_sha=NEW_HEAD,
+            approved_head_sha=NEW_HEAD,
+        )
+        self.assertTrue(result["approval_usable"])
+
+    def test_stale_reviewer_lease_cannot_cross_heads(self):
+        result = lease_usable_at_head(
+            lease_kind="reviewer",
+            lease_head_sha=PR_HEAD,
+            current_head_sha=NEW_HEAD,
+        )
+        self.assertFalse(result["lease_usable"])
+
+    def test_stale_merger_lease_cannot_cross_heads(self):
+        result = lease_usable_at_head(
+            lease_kind="merger",
+            lease_head_sha=PR_HEAD,
+            current_head_sha=NEW_HEAD,
+        )
+        self.assertFalse(result["lease_usable"])
+
+
+class TestSequentialQueue(unittest.TestCase):
+    def test_reassess_after_merge_and_master_refresh(self):
+        result = assess_sequential_queue_step(
+            just_merged=True,
+            master_refreshed=True,
+            next_pr_number=101,
+        )
+        self.assertTrue(result["reassess_allowed"])
+        self.assertTrue(result["process_next"])
+        self.assertTrue(result["post_merge_cleanup_handoff"])
+        self.assertEqual(result["next_pr_number"], 101)
+
+    def test_block_reassess_without_master_refresh(self):
+        result = assess_sequential_queue_step(
+            just_merged=True,
+            master_refreshed=False,
+            next_pr_number=101,
+        )
+        self.assertFalse(result["reassess_allowed"])
+        self.assertTrue(any("master must be refreshed" in r for r in result["reasons"]))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_resolve_task_capability.py

@@ -231,9 +231,16 @@ class TestResolveTaskCapability(unittest.TestCase):
             self.assertIn("gitea.issue.comment", res["active_profile_allowed_operations"])
 
     def test_resolve_unknown_task_fails_closed(self):
+        # #723: unknown tasks return structured unknown_task (no ValueError escape).
         with patch.dict(os.environ, self._env("author-profile")):
-            with self.assertRaises(ValueError):
-                mcp_server.gitea_resolve_task_capability(task="invalid_task_xyz", remote="prgs")
+            res = mcp_server.gitea_resolve_task_capability(
+                task="invalid_task_xyz", remote="prgs"
+            )
+            self.assertIsInstance(res, dict)
+            self.assertEqual(res.get("reason_code"), "unknown_task", res)
+            self.assertFalse(res.get("allowed_in_current_session"))
+            self.assertTrue(res.get("stop_required"))
+            self.assertIs(res.get("mutation_performed"), False)
 
     # Additional regression tests per #145 for permission boundaries and structured guidance
     def test_issue_comment_does_not_imply_close(self):
@@ -439,8 +446,11 @@ class TestResolveTaskCapability(unittest.TestCase):
             res = mcp_server.gitea_resolve_task_capability(task="close_pr", remote="prgs")
             self.assertEqual(res["required_operation_permission"], "gitea.pr.close")
             for unknown in ("close_pull_request", "close", "pr_close"):
-                with self.assertRaises(ValueError):
-                    mcp_server.gitea_resolve_task_capability(task=unknown, remote="prgs")
+                unk = mcp_server.gitea_resolve_task_capability(
+                    task=unknown, remote="prgs"
+                )
+                self.assertEqual(unk.get("reason_code"), "unknown_task", unk)
+                self.assertFalse(unk.get("allowed_in_current_session"))
 
 if __name__ == "__main__":
     unittest.main()

tests/test_review_drafts.py

@@ -68,8 +68,6 @@ class TestReviewDrafts(unittest.TestCase):
 
     @patch("gitea_mcp_server.api_request")
     def test_save_draft_success(self, mock_api):
-        print("GITEA_MCP_SERVER FILE:", gitea_mcp_server.__file__)
-        print("DIR OF GITEA_MCP_SERVER:", dir(gitea_mcp_server))
         mock_api.return_value = {
             "head": {"sha": "headsha1234567890"},
             "base": {"ref": "master", "sha": "basesha0987654321"},
@@ -285,3 +283,104 @@ class TestReviewDrafts(unittest.TestCase):
         )
         self.assertFalse(res.get("success"))
         self.assertIn("worktree binding mismatch", res.get("reasons")[0])
+
+    @patch("gitea_mcp_server.api_request")
+    @patch("gitea_mcp_server._fetch_pr_comments")
+    def test_resume_draft_foreign_lease(self, mock_comments, mock_api):
+        mock_api.return_value = {
+            "state": "open",
+            "head": {"sha": "headsha1234567890"},
+            "base": {"ref": "master", "sha": "basesha0987654321"},
+        }
+        gitea_mcp_server.gitea_save_review_draft(
+            pr_number=587,
+            action="approve",
+            body="Good changes",
+            expected_head_sha="headsha1234567890",
+            remote="prgs",
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+            worktree_path="/tmp/test-worktree",
+        )
+        reviewer_pr_lease.record_session_lease({
+            "pr_number": 587,
+            "session_id": "my-session",
+        })
+        mock_comments.return_value = [
+            {
+                "id": 9,
+                "user": {"username": "sysadmin"},
+                "body": (
+                    "<!-- mcp-review-lease:v1 -->\n"
+                    "repo: Scaled-Tech-Consulting/Gitea-Tools\n"
+                    "pr: #587\n"
+                    "reviewer_identity: sysadmin\n"
+                    "profile: prgs-reviewer\n"
+                    "session_id: foreign-session-999\n"
+                    "phase: claimed\n"
+                ),
+            }
+        ]
+        res = gitea_mcp_server.gitea_resume_review_draft(
+            pr_number=587,
+            remote="prgs",
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+            worktree_path="/tmp/test-worktree",
+        )
+        self.assertFalse(res.get("success"))
+        self.assertTrue(
+            any("foreign-session-999" in r or "session_id" in r for r in (res.get("reasons") or []))
+        )
+
+    @patch("gitea_mcp_server.api_request")
+    @patch("gitea_mcp_server._fetch_pr_comments")
+    @patch("gitea_mcp_server.terminal_review_hard_stop_reasons")
+    def test_resume_draft_terminal_lock_blocks(self, mock_hard_stop, mock_comments, mock_api):
+        mock_api.return_value = {
+            "state": "open",
+            "head": {"sha": "headsha1234567890"},
+            "base": {"ref": "master", "sha": "basesha0987654321"},
+        }
+        gitea_mcp_server.gitea_save_review_draft(
+            pr_number=587,
+            action="approve",
+            body="Good changes",
+            expected_head_sha="headsha1234567890",
+            remote="prgs",
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+            worktree_path="/tmp/test-worktree",
+        )
+        reviewer_pr_lease.record_session_lease({
+            "pr_number": 587,
+            "session_id": "session-1234",
+        })
+        mock_comments.return_value = [
+            {
+                "id": 1,
+                "user": {"username": "sysadmin"},
+                "body": (
+                    "<!-- mcp-review-lease:v1 -->\n"
+                    "repo: Scaled-Tech-Consulting/Gitea-Tools\n"
+                    "pr: #587\n"
+                    "reviewer_identity: sysadmin\n"
+                    "profile: prgs-reviewer\n"
+                    "session_id: session-1234\n"
+                    "phase: claimed\n"
+                ),
+            }
+        ]
+        mock_hard_stop.return_value = [
+            "terminal review mutation already recorded for PR #587 (approve); #332 hard-stop"
+        ]
+        res = gitea_mcp_server.gitea_resume_review_draft(
+            pr_number=587,
+            remote="prgs",
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+            worktree_path="/tmp/test-worktree",
+        )
+        self.assertFalse(res.get("success"))
+        self.assertTrue(any("#332" in r or "terminal" in r for r in (res.get("reasons") or [])))
+

tests/test_sentry_observability.py

@@ -0,0 +1,412 @@
+"""Tests for optional self-hosted Sentry observability (#606).
+
+Covers the pure module (config, redaction, event/check-in builders) and the
+runtime capture paths using a fake ``sentry_sdk``, so nothing ever touches the
+network. Critically proves the feature is a no-op when disabled or DSN-less
+(acceptance criterion 1) and that secrets/paths/session-state are never sent
+(criterion 5).
+"""
+
+import sys
+import contextlib
+from pathlib import Path
+
+import pytest
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+import sentry_observability as so  # noqa: E402
+
+
+# ── Fake SDK ────────────────────────────────────────────────────────────────
+class _FakeScope:
+    def __init__(self):
+        self.tags = {}
+        self.extras = {}
+
+    def set_tag(self, k, v):
+        self.tags[k] = v
+
+    def set_extra(self, k, v):
+        self.extras[k] = v
+
+
+class FakeSentrySDK:
+    """Minimal stand-in exposing the SDK surface sentry_observability uses."""
+
+    def __init__(self):
+        self.init_kwargs = None
+        self.events = []
+        self.exceptions = []
+        self.checkins = []
+        self.last_scope = None
+
+    def init(self, **kwargs):
+        self.init_kwargs = kwargs
+
+    def capture_event(self, event):
+        self.events.append(event)
+
+    def capture_exception(self, exc):
+        self.exceptions.append(exc)
+
+    def capture_checkin(self, **payload):
+        self.checkins.append(payload)
+
+    @contextlib.contextmanager
+    def push_scope(self):
+        self.last_scope = _FakeScope()
+        yield self.last_scope
+
+
+@pytest.fixture(autouse=True)
+def _reset_state():
+    """Isolate module init state between tests."""
+    so.reset_for_tests()
+    yield
+    so.reset_for_tests()
+
+
+@pytest.fixture
+def fake_sdk(monkeypatch):
+    sdk = FakeSentrySDK()
+    monkeypatch.setattr(so, "sentry_sdk", sdk)
+    return sdk
+
+
+# ── Config / gating ─────────────────────────────────────────────────────────
+def test_disabled_by_default_empty_env():
+    cfg = so.load_config(env={})
+    assert cfg.enabled is False
+    assert cfg.active is False
+
+
+def test_enabled_flag_without_dsn_is_not_active():
+    cfg = so.load_config(env={"MCP_SENTRY_ENABLED": "1"})
+    assert cfg.enabled is True
+    assert cfg.dsn is None
+    assert cfg.active is False  # DSN required
+
+
+def test_dsn_without_enabled_flag_is_not_active():
+    cfg = so.load_config(env={"SENTRY_DSN": "https://[email protected]/1"})
+    assert cfg.active is False
+
+
+def test_active_requires_enabled_and_dsn():
+    cfg = so.load_config(
+        env={"MCP_SENTRY_ENABLED": "true", "SENTRY_DSN": "https://[email protected]/1"}
+    )
+    assert cfg.active is True
+
+
+def test_truthy_variants():
+    for val in ("1", "true", "YES", "On"):
+        cfg = so.load_config(env={"MCP_SENTRY_ENABLED": val})
+        assert cfg.enabled is True
+    for val in ("0", "false", "no", "", "off"):
+        cfg = so.load_config(env={"MCP_SENTRY_ENABLED": val})
+        assert cfg.enabled is False
+
+
+def test_traces_sample_rate_parsed_and_clamped():
+    assert so.load_config(env={"MCP_SENTRY_TRACES_SAMPLE_RATE": "0.25"}).traces_sample_rate == 0.25
+    assert so.load_config(env={"MCP_SENTRY_TRACES_SAMPLE_RATE": "5"}).traces_sample_rate == 1.0
+    assert so.load_config(env={"MCP_SENTRY_TRACES_SAMPLE_RATE": "-1"}).traces_sample_rate == 0.0
+    assert so.load_config(env={"MCP_SENTRY_TRACES_SAMPLE_RATE": "junk"}).traces_sample_rate == 0.0
+
+
+def test_safe_summary_has_no_dsn_value():
+    cfg = so.load_config(
+        env={"MCP_SENTRY_ENABLED": "1", "SENTRY_DSN": "https://[email protected]/1"}
+    )
+    summary = cfg.safe_summary()
+    assert summary["dsn_present"] is True
+    assert "secret" not in repr(summary)
+    assert "dsn" not in summary  # only presence, never the value
+
+
+# ── init_sentry ─────────────────────────────────────────────────────────────
+def test_init_noop_when_disabled(fake_sdk):
+    status = so.init_sentry(so.load_config(env={}))
+    assert status["initialized"] is False
+    assert fake_sdk.init_kwargs is None  # no SDK init
+    assert so.is_initialized() is False
+
+
+def test_init_noop_when_enabled_but_missing_dsn(fake_sdk):
+    status = so.init_sentry(so.load_config(env={"MCP_SENTRY_ENABLED": "1"}))
+    assert status["initialized"] is False
+    assert fake_sdk.init_kwargs is None
+
+
+def test_init_reports_missing_sdk(monkeypatch):
+    monkeypatch.setattr(so, "sentry_sdk", None)
+    status = so.init_sentry(
+        so.load_config(
+            env={"MCP_SENTRY_ENABLED": "1", "SENTRY_DSN": "https://[email protected]/1"}
+        )
+    )
+    assert status["initialized"] is False
+    assert "not installed" in status["reason"]
+
+
+def test_init_configures_sdk_with_scrubber(fake_sdk):
+    status = so.init_sentry(
+        so.load_config(
+            env={
+                "MCP_SENTRY_ENABLED": "1",
+                "SENTRY_DSN": "https://[email protected]/1",
+                "SENTRY_ENVIRONMENT": "prod",
+                "MCP_SENTRY_TRACES_SAMPLE_RATE": "0.1",
+            }
+        )
+    )
+    assert status["initialized"] is True
+    assert so.is_initialized() is True
+    kw = fake_sdk.init_kwargs
+    assert kw["dsn"] == "https://[email protected]/1"
+    assert kw["environment"] == "prod"
+    assert kw["traces_sample_rate"] == 0.1
+    assert kw["before_send"] is so.scrub_event
+    assert kw["send_default_pii"] is False
+
+
+def test_init_enable_logs_wires_log_scrubber(fake_sdk):
+    so.init_sentry(
+        so.load_config(
+            env={
+                "MCP_SENTRY_ENABLED": "1",
+                "SENTRY_DSN": "https://[email protected]/1",
+                "MCP_SENTRY_ENABLE_LOGS": "1",
+            }
+        )
+    )
+    exp = fake_sdk.init_kwargs["_experiments"]
+    assert exp["enable_logs"] is True
+    assert exp["before_send_log"] is so.scrub_event
+
+
+def test_init_never_raises_on_sdk_failure(monkeypatch):
+    class Boom:
+        def init(self, **kwargs):
+            raise RuntimeError("sentry down")
+
+    monkeypatch.setattr(so, "sentry_sdk", Boom())
+    status = so.init_sentry(
+        so.load_config(
+            env={"MCP_SENTRY_ENABLED": "1", "SENTRY_DSN": "https://[email protected]/1"}
+        )
+    )
+    assert status["initialized"] is False
+    assert "init failed" in status["reason"]
+
+
+# ── Redaction (fail closed) ─────────────────────────────────────────────────
+def test_build_tags_allowlist_only():
+    tags = so.build_tags(role="author", secret_thing="leak", pid=123)
+    assert tags["role"] == "author"
+    assert tags["pid"] == "123"
+    assert "secret_thing" not in tags
+
+
+def test_build_tags_hashes_session_id():
+    tags = so.build_tags(session_id="prgs-author-20479-cf9ac178")
+    assert "session_id" not in tags
+    assert "session_id_hash" in tags
+    assert tags["session_id_hash"] != "prgs-author-20479-cf9ac178"
+    assert len(tags["session_id_hash"]) == 12
+
+
+def test_build_tags_collapses_worktree_path():
+    tags = so.build_tags(
+        worktree_path="/Users/x/Development/Gitea-Tools/branches/issue-606-sentry-observability"
+    )
+    assert "worktree_path" not in tags
+    assert tags["worktree_category"] == "author"
+
+
+def test_build_tags_drops_value_that_scrubs_to_redacted():
+    # A tag value that is itself a token gets scrubbed then dropped.
+    tags = so.build_tags(capability="token abcdef1234567890")
+    assert "capability" not in tags
+
+
+def test_sanitize_path_categories():
+    assert so.sanitize_path("/repo/branches/review-pr-654") == "reviewer"
+    assert so.sanitize_path("/repo/branches/merge-pr-1") == "merger"
+    assert so.sanitize_path("/repo/branches/reconcile-pr-1") == "reconciler"
+    assert so.sanitize_path("/repo/branches/feat-issue-606") == "author"
+    assert so.sanitize_path("/x/y/Gitea-Tools") == "root"
+
+
+def test_redact_value_scrubs_secrets_and_paths():
+    out = so.redact_value(
+        {
+            "token": "abc123",
+            "note": "Authorization: Bearer sk_live_abcdefgh12345",
+            "path": "/Users/jasonwalker/Development/Gitea-Tools/secret",
+            "dsn": "https://[email protected]/1",
+            "safe": "hello",
+        }
+    )
+    assert out["token"] == so.REDACTED
+    assert out["dsn"] == so.REDACTED
+    assert "sk_live" not in out["note"]
+    assert so.REDACTED_PATH in out["path"]
+    assert "/Users/" not in out["path"]
+    assert out["safe"] == "hello"
+
+
+def test_redact_value_forbidden_prompt_and_session_state():
+    out = so.redact_value(
+        {"prompt": "full body", "session_state": "{...}", "keep": "ok"}
+    )
+    assert out["prompt"] == so.REDACTED
+    assert out["session_state"] == so.REDACTED
+    assert out["keep"] == "ok"
+
+
+def test_scrub_event_redacts_nested_and_drops_server_name():
+    event = {
+        "server_name": "some-host",
+        "message": "boom",
+        "extra": {"token": "leak", "ok": "1"},
+    }
+    scrubbed = so.scrub_event(event)
+    assert "server_name" not in scrubbed
+    assert scrubbed["extra"]["token"] == so.REDACTED
+    assert scrubbed["extra"]["ok"] == "1"
+
+
+def test_scrub_event_drops_non_dict():
+    assert so.scrub_event("not a dict") is None
+    assert so.scrub_event(None) is None
+
+
+# ── Event builders ──────────────────────────────────────────────────────────
+def test_build_blocker_event_structure_and_next_action():
+    event = so.build_blocker_event(
+        "active_foreign_lease",
+        message="blocked by foreign lease",
+        next_action="wait or adopt via allocator",
+        level="warning",
+        tags={"pr_number": 606, "session_id": "s-123"},
+    )
+    assert event["tags"]["blocker_type"] == "active_foreign_lease"
+    assert event["tags"]["pr_number"] == "606"
+    assert "session_id" not in event["tags"]
+    assert event["tags"]["session_id_hash"]
+    assert event["extra"]["canonical_next_action"] == "wait or adopt via allocator"
+    assert event["fingerprint"] == ["workflow-blocker", "active_foreign_lease"]
+
+
+def test_build_blocker_event_invalid_level_defaults_warning():
+    event = so.build_blocker_event("x", level="nonsense")
+    assert event["level"] == "warning"
+
+
+def test_build_checkin_payload_maps_slugs():
+    for key, slug in so.MONITOR_SLUGS.items():
+        payload = so.build_checkin_payload(key, "ok")
+        assert payload["monitor_slug"] == slug
+        assert payload["status"] == "ok"
+
+
+def test_build_checkin_payload_explicit_slug_passthrough():
+    payload = so.build_checkin_payload("custom-slug", "in_progress", duration=1.5)
+    assert payload["monitor_slug"] == "custom-slug"
+    assert payload["duration"] == 1.5
+
+
+def test_build_checkin_payload_rejects_bad_status():
+    with pytest.raises(ValueError):
+        so.build_checkin_payload("allocator_health", "bogus")
+
+
+def test_all_six_monitors_registered():
+    assert set(so.MONITOR_SLUGS) == {
+        "stale_lease_scan",
+        "terminal_lock_scan",
+        "allocator_health",
+        "namespace_health",
+        "dashboard_freshness",
+        "reconciler_cleanup",
+    }
+
+
+# ── Capture paths (fail open) ───────────────────────────────────────────────
+def test_capture_workflow_blocker_noop_when_disabled(fake_sdk):
+    # not initialised
+    event = so.capture_workflow_blocker("some_blocker", message="x")
+    assert isinstance(event, dict)  # still returns redacted event
+    assert fake_sdk.events == []  # but nothing sent
+
+
+def test_capture_workflow_blocker_sends_when_initialised(fake_sdk):
+    so.init_sentry(
+        so.load_config(
+            env={"MCP_SENTRY_ENABLED": "1", "SENTRY_DSN": "https://[email protected]/1"}
+        )
+    )
+    so.capture_workflow_blocker("terminal_lock_occupied", message="held")
+    assert len(fake_sdk.events) == 1
+    assert fake_sdk.events[0]["tags"]["blocker_type"] == "terminal_lock_occupied"
+
+
+def test_capture_exception_noop_when_disabled(fake_sdk):
+    assert so.capture_exception(ValueError("x")) is False
+    assert fake_sdk.exceptions == []
+
+
+def test_capture_exception_sends_scrubbed_tags(fake_sdk):
+    so.init_sentry(
+        so.load_config(
+            env={"MCP_SENTRY_ENABLED": "1", "SENTRY_DSN": "https://[email protected]/1"}
+        )
+    )
+    ok = so.capture_exception(
+        RuntimeError("bad"), tags={"mutation_tool": "gitea_merge_pr", "leaky": "x"}
+    )
+    assert ok is True
+    assert len(fake_sdk.exceptions) == 1
+    assert fake_sdk.last_scope.tags["mutation_tool"] == "gitea_merge_pr"
+    assert "leaky" not in fake_sdk.last_scope.tags
+
+
+def test_capture_exception_never_raises(monkeypatch, fake_sdk):
+    so.init_sentry(
+        so.load_config(
+            env={"MCP_SENTRY_ENABLED": "1", "SENTRY_DSN": "https://[email protected]/1"}
+        )
+    )
+
+    def boom(exc):
+        raise RuntimeError("sdk exploded")
+
+    monkeypatch.setattr(fake_sdk, "capture_exception", boom)
+    # Must swallow the SDK failure (fail open).
+    assert so.capture_exception(ValueError("y")) is False
+
+
+def test_monitor_checkin_noop_when_disabled(fake_sdk):
+    payload = so.monitor_checkin("allocator_health", "ok")
+    assert payload["monitor_slug"] == "gitea-mcp-allocator-health"
+    assert fake_sdk.checkins == []  # not sent while disabled
+
+
+def test_monitor_checkin_sends_when_initialised(fake_sdk):
+    so.init_sentry(
+        so.load_config(
+            env={"MCP_SENTRY_ENABLED": "1", "SENTRY_DSN": "https://[email protected]/1"}
+        )
+    )
+    so.monitor_checkin("stale_lease_scan", "ok")
+    assert fake_sdk.checkins == [
+        {"monitor_slug": "gitea-mcp-stale-lease-scan", "status": "ok"}
+    ]
+
+
+def test_monitor_checkin_invalid_status_returns_none(fake_sdk):
+    assert so.monitor_checkin("allocator_health", "bogus") is None
+    assert fake_sdk.checkins == []

tests/test_stale_review_decision_lock_cleanup.py

@@ -188,20 +188,32 @@ class TestActiveHardStopPreserved(unittest.TestCase):
         mcp_server._save_review_decision_lock(None)
         mcp_server.review_workflow_load.clear_review_workflow_load()
 
-    def test_open_approve_still_blocks_other_pr_mark_ready(self):
-        _seed([APPROVED_OPEN])
-        reasons = mcp_server.terminal_review_hard_stop_reasons(592, "mark_ready")
-        self.assertTrue(reasons)
-        self.assertIn("#332", reasons[0])
-        self.assertIn("gitea_cleanup_stale_review_decision_lock", reasons[0])
+    def test_open_approve_blocks_same_pr_not_other_pr_mark_ready(self):
+        _seed([APPROVED_OPEN])  # approve on PR #700
+        # Same PR still hard-stopped for mark_ready.
+        reasons_same = mcp_server.terminal_review_hard_stop_reasons(
+            700, "mark_ready"
+        )
+        self.assertTrue(reasons_same)
+        self.assertIn("#332", reasons_same[0])
+        # #693: other PR may mark_ready without cleanup.
+        self.assertEqual(
+            mcp_server.terminal_review_hard_stop_reasons(592, "mark_ready"),
+            [],
+        )
 
-    def test_request_changes_still_blocks_everything(self):
-        _seed([RC_OPEN])
+    def test_request_changes_still_blocks_same_pr(self):
+        _seed([RC_OPEN])  # request_changes on PR #701
         for op in ("merge", "mark_ready", "review"):
             self.assertTrue(
-                mcp_server.terminal_review_hard_stop_reasons(592, op),
+                mcp_server.terminal_review_hard_stop_reasons(701, op),
                 msg=op,
             )
+        # Foreign PR review path is open (#693).
+        self.assertEqual(
+            mcp_server.terminal_review_hard_stop_reasons(592, "mark_ready"),
+            [],
+        )
 
 
 # --------------------------------------------------------------------------- #

tests/test_task_capability_role_invariants.py

@@ -19,12 +19,7 @@ import unittest
 
 import gitea_config
 from role_session_router import MERGER_TASKS, REVIEWER_TASKS
-from task_capability_map import (
-    ROLE_EXCLUSIVE_TASKS,
-    TASK_CAPABILITY_MAP,
-    required_permission,
-    required_role,
-)
+from task_capability_map import required_permission, required_role
 
 # Canonical role-profile permission shape. Mirrors the configured
 # author/reviewer/merger/reconciler profiles (profiles.json v2 role split):
@@ -206,44 +201,5 @@ class TestMergerBoundary(unittest.TestCase):
                 )
 
 
-class TestRoleExclusiveSetIntegrity(unittest.TestCase):
-    """#723 AC1/AC5: the shared role-exclusive set stays coherent."""
-
-    def test_every_role_exclusive_task_exists_in_capability_map(self):
-        for task in sorted(ROLE_EXCLUSIVE_TASKS):
-            with self.subTest(task=task):
-                self.assertIn(
-                    task,
-                    TASK_CAPABILITY_MAP,
-                    f"role-exclusive task {task!r} missing from the "
-                    f"capability map — required_role() would raise and the "
-                    f"resolver would 500 instead of failing closed",
-                )
-
-    def test_formal_review_tasks_are_role_exclusive(self):
-        for task in FORMAL_REVIEW_TASKS:
-            with self.subTest(task=task):
-                self.assertIn(task, ROLE_EXCLUSIVE_TASKS)
-
-    def test_merge_pr_is_role_exclusive_and_merger_satisfiable(self):
-        """AC1 merger equivalent: merging must stay possible for a canonical
-        merger profile (permission AND role together)."""
-        self.assertIn("merge_pr", ROLE_EXCLUSIVE_TASKS)
-        self.assertTrue(
-            _profile_satisfies("merger", "merge_pr"),
-            "no canonical merger profile satisfies merge_pr — merging would "
-            "be impossible for every configured profile",
-        )
-
-    def test_reconciler_cleanup_tasks_stay_reconciler_satisfiable(self):
-        for task in ("cleanup_merged_pr_branch", "reconciliation_cleanup"):
-            with self.subTest(task=task):
-                self.assertIn(task, ROLE_EXCLUSIVE_TASKS)
-                self.assertTrue(
-                    _profile_satisfies("reconciler", task),
-                    f"no canonical reconciler profile satisfies {task!r}",
-                )
-
-
 if __name__ == "__main__":
     unittest.main()

tests/test_terminal_review_hard_stop.py

@@ -1,10 +1,10 @@
-"""Tests for the session hard-stop after a terminal review mutation (#332).
+"""Tests for the session hard-stop after a terminal review mutation (#332/#693).
 
 Extends the single-terminal-decision machinery (#211): after a live
-REQUEST_CHANGES the run must stop; after an APPROVE only the merge sequence
-for that same PR may continue; a different PR can never be reviewed or
-merged in the same run; duplicate REQUEST_CHANGES at an unchanged head is
-suppressed.
+REQUEST_CHANGES on a PR head the same-head path must stop; after an APPROVE
+only the merge sequence for that same PR may continue; #693 allows a
+*different* PR to be mark_ready/review in the same durable lock (cross-PR
+isolation); duplicate REQUEST_CHANGES at an unchanged head is suppressed.
 """
 import os
 import unittest
@@ -15,7 +15,10 @@ import mcp_server
 HEAD_SHA = "a" * 40
 
 
-def _lock(mutations=None, correction=False):
+def _lock(mutations=None, correction=False, correction_pr=None):
+    mutations = list(mutations or [])
+    if correction and correction_pr is None and mutations:
+        correction_pr = mutations[-1].get("pr_number")
     return {
         "task": "review_pr",
         "remote": "prgs",
@@ -29,9 +32,11 @@ def _lock(mutations=None, correction=False):
         "ready_remote": None,
         "ready_org": None,
         "ready_repo": None,
-        "live_mutations": list(mutations or []),
+        "live_mutations": mutations,
         "correction_authorized": correction,
-        "correction_reason": None,
+        "correction_reason": "test" if correction else None,
+        "correction_pr_number": correction_pr if correction else None,
+        "correction_head_sha": None,
     }
 
 
@@ -64,25 +69,35 @@ class TestTerminalHardStopReasons(unittest.TestCase):
         self.assertEqual(
             mcp_server.terminal_review_hard_stop_reasons(5, "merge"), [])
 
-    def test_request_changes_blocks_everything(self):
+    def test_request_changes_blocks_same_pr_allows_other_pr_review(self):
         _seed([RC_A])
+        # Same PR: still hard-stopped for mark_ready/review/merge.
         for op in ("merge", "mark_ready", "review"):
-            for pr in (5, 6):
-                reasons = mcp_server.terminal_review_hard_stop_reasons(pr, op)
-                self.assertTrue(reasons, f"{op}/{pr}")
-                self.assertIn("request_changes on PR #5", reasons[0])
-                self.assertIn("#332", reasons[0])
+            reasons = mcp_server.terminal_review_hard_stop_reasons(5, op)
+            self.assertTrue(reasons, f"{op}/5")
+            self.assertIn("request_changes on PR #5", reasons[0])
+            self.assertIn("#332", reasons[0])
+        # #693: different PR may mark_ready / review (not merge via hard-stop alone).
+        self.assertEqual(
+            mcp_server.terminal_review_hard_stop_reasons(6, "mark_ready"), [])
+        self.assertEqual(
+            mcp_server.terminal_review_hard_stop_reasons(6, "review"), [])
+        # Merge of an unapproved other PR remains blocked by hard-stop path
+        # (last terminal is not approve of PR 6).
+        self.assertTrue(
+            mcp_server.terminal_review_hard_stop_reasons(6, "merge"))
 
-    def test_approve_allows_same_pr_merge_only(self):
+    def test_approve_allows_same_pr_merge_and_other_pr_review(self):
         _seed([APPROVED_A])
         self.assertEqual(
             mcp_server.terminal_review_hard_stop_reasons(5, "merge"), [])
         self.assertTrue(
             mcp_server.terminal_review_hard_stop_reasons(6, "merge"))
-        self.assertTrue(
-            mcp_server.terminal_review_hard_stop_reasons(6, "mark_ready"))
-        self.assertTrue(
-            mcp_server.terminal_review_hard_stop_reasons(6, "review"))
+        # #693 cross-PR isolation: other PR formal review path is open.
+        self.assertEqual(
+            mcp_server.terminal_review_hard_stop_reasons(6, "mark_ready"), [])
+        self.assertEqual(
+            mcp_server.terminal_review_hard_stop_reasons(6, "review"), [])
 
     def test_correction_reopens_review_path_only(self):
         _seed([RC_A], correction=True)
@@ -90,9 +105,11 @@ class TestTerminalHardStopReasons(unittest.TestCase):
             mcp_server.terminal_review_hard_stop_reasons(5, "mark_ready"), [])
         self.assertEqual(
             mcp_server.terminal_review_hard_stop_reasons(5, "review"), [])
-        # correction never opens a cross-PR merge
+        # scoped correction never opens a cross-PR merge
         self.assertTrue(
             mcp_server.terminal_review_hard_stop_reasons(6, "merge"))
+        # unscoped/cross-PR correction must not lift PR 6 via correction alone
+        # (PR 6 is allowed by isolation, not by correction scope)
 
 
 class TestMergeHardStopWiring(unittest.TestCase):
@@ -126,14 +143,29 @@ class TestMarkFinalHardStopWiring(unittest.TestCase):
         mcp_server._save_review_decision_lock(None)
         mcp_server.review_workflow_load.clear_review_workflow_load()
 
-    def test_mark_ready_blocked_after_terminal_mutation(self):
+    def test_mark_ready_same_pr_blocked_after_terminal_mutation(self):
         _seed([RC_A])
         result = mcp_server.gitea_mark_final_review_decision(
-            pr_number=6, action="approve", remote="prgs")
+            pr_number=5, action="approve", remote="prgs",
+            expected_head_sha=HEAD_SHA)
         self.assertFalse(result["marked_ready"])
         self.assertTrue(
             any("#332" in r for r in result["reasons"]), result["reasons"])
 
+    def test_mark_ready_other_pr_allowed_after_foreign_terminal(self):
+        """#693: foreign open-PR terminal must not block mark_final on PR B."""
+        _seed([RC_A])
+        no_lease = {"block": False, "reasons": [], "mutation_allowed": True}
+        with patch("mcp_server._list_pr_lease_comments", return_value=[]), \
+                patch("mcp_server._pr_work_lease_reviewer_block",
+                      return_value=no_lease), \
+                patch.object(mcp_server, "gitea_check_pr_eligibility",
+                             return_value={"head_sha": HEAD_SHA, "eligible": True}):
+            result = mcp_server.gitea_mark_final_review_decision(
+                pr_number=6, action="approve", remote="prgs",
+                expected_head_sha=HEAD_SHA)
+        self.assertTrue(result["marked_ready"], result.get("reasons"))
+
 
 def _feedback(blocking, stale=False, success=True):
     return {