Merge branch 'master' into fix/issue-702-stale-binding-lease-recovery

Reviewed-on: #680
Reviewed-by: sysadmin <[email protected]>

Operator break-glass merge record

PR #680 was independently assessed at head `7c77df0c527f06ae4196e010e65537d361805a7b`.

* Gitea reports the PR open and mergeable.
* The focused anti-stomp suite passed: 43 tests.
* The technical review found the change approve-worthy.
* The previous approval is stale because it applies to an earlier head.
* A fresh formal approval could not be submitted because every MCP namespace remained bound to pre-#694 startup SHA `f65069d…`, while live master is `3e78db5…`.

This is an operator-authorized break-glass merge caused by stale MCP runtime parity, not an unresolved code finding. Proceeding with **Create merge commit** only if the PR head remains exactly `7c77df0c527f06ae4196e010e65537d361805a7b`, Gitea still reports it mergeable, and no new blocking checks or feedback appear.

.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/architecture/mcp-stable-control-runtime-policy-adr.md

@@ -0,0 +1,198 @@
+# ADR: Stable control runtime vs dev runtime (Gitea MCP)
+
+- **Status:** Accepted (policy effective immediately for LLM sessions; tooling may lag)
+- **Date:** 2026-07-09
+- **Tracking issue:** [#615](https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/issues/615)
+- **Related:**
+  - [#543](https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/issues/543) / `docs/mcp-namespace-health.md` — client-namespace health
+  - `docs/mcp-namespace-eof-recovery.md` — reconnect-only EOF recovery (no PID kill)
+  - [#558](https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/issues/558) / `docs/mcp-daemon-import-guard.md` — sanctioned daemon
+  - [#557](https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/issues/557) / `docs/bootstrap-review-path.md` — controller bootstrap for self-hosted fixes
+  - Allocator / control-plane ADR: `docs/architecture/mcp-allocator-control-plane-observability-adr.md` (#613 / PR #614)
+
+## 1. Context
+
+The Gitea MCP server is the **control plane** for real issue/PR mutations (create, comment, lock, review, merge, etc.). When author/reviewer/merger/reconciler sessions kill or restart that process, relaunch it from a feature worktree, or edit the checkout that process loads, operators observe:
+
+- Mid-session identity/preflight resets
+- Stale-runtime vs master parity failures
+- IDE transport EOF / “tool not found” while code on disk has changed
+- Accidental production mutations from experimental code
+
+This ADR separates **stable control runtime** from **dev/test runtime** and defines promotion proof.
+
+## 2. Decision
+
+### 2.1 Stable control runtime
+
+The Gitea MCP server used for **real workflow mutations** is the **stable control runtime**.
+
+Characteristics:
+
+- Loads a known, promoted revision of Gitea-Tools (or the packaged release layout operators designate)
+- Registered in the IDE/client as the production namespaces (`gitea-tools`, `gitea-reviewer`, `gitea-merger`, `gitea-reconciler`, etc.)
+- Holds production profile credentials via sanctioned keychain/env paths only
+
+### 2.2 Dev / test runtime
+
+MCP **server code** development and testing:
+
+- Happens in isolated **`branches/`** worktrees (or other non-stable checkouts)
+- May use a **separate** dev/test MCP runtime/process when process-level testing is required
+- **Must not** be used for real Gitea mutations on production issues/PRs
+
+### 2.3 Forbidden actions (normal sessions)
+
+Normal **author, reviewer, merger, and reconciler** LLM sessions **must not**:
+
+| Forbidden | Why |
+|-----------|-----|
+| Kill the running MCP server process | Drops all concurrent sessions; loses preflight state |
+| Restart / relaunch the MCP server process | Same as kill; causes stale/identity churn mid-workflow |
+| Relaunch MCP from a development worktree | Runs unpromoted code against production mutations |
+| Edit files in the stable runtime checkout | Hot-mutates control plane under concurrent users |
+| Use experimental/dev MCP for real Gitea mutations | Bypasses promotion proof and audit expectations |
+| Bypass or self-reset a stale master-parity gate | The gate is fail-closed; only an operator reload restores parity |
+
+**LLM-allowed vs operator-owned (authoritative split):**
+
+| Actor | May do | Must not do |
+|-------|--------|-------------|
+| **LLM session** | Call tools on the already-running stable namespaces; **client reconnect** after transport EOF (no process kill); pass `worktree_path` / role worktree args; report blockers and stop mutations when unhealthy/stale | Kill, restart, or relaunch any MCP process; bump config mtimes to force reload; edit the stable checkout; switch to a dev MCP for production mutations |
+| **Operator / release-manager** | Supervised restart/reload of the **stable** control runtime; dual-namespace client configuration; §2.4 promotions; incident recovery | — |
+
+EOF / transport recovery for LLM sessions: **client reconnect only** (see `docs/mcp-namespace-eof-recovery.md`). Do not “fix” health by killing PIDs or bumping MCP config mtimes as a normal session procedure.
+
+This ADR **supersedes** any older runbook wording that told the LLM to relaunch or restart the client/MCP as a self-service step. Where `docs/llm-workflow-runbooks.md` (or wiki runbooks) discuss dual-namespace setup or workspace rebind, **process restart/relaunch is operator-owned**; the LLM stops, reports, and waits.
+
+### 2.4 Promotion (operator / release-manager only)
+
+Promotion of a new revision into the stable control runtime is an **explicit operator/release-manager action**, not an LLM self-service step.
+
+A promotion **must record** (issue comment, release note, or promotion ledger):
+
+| Field | Description |
+|-------|-------------|
+| **previous runtime SHA** | Commit previously loaded by stable runtime |
+| **promoted runtime SHA** | Commit after promotion |
+| **source branch/PR** | Where the change was reviewed |
+| **restart/reload method** | How the process was cycled (e.g. supervised restart, client reload) |
+| **health check proof** | Client-namespace probe success (`gitea_whoami` / namespace health) |
+| **identity/profile proof** | Expected profile(s) and username(s) after reload |
+| **workspace/root proof** | Stable checkout path / root matches intended layout |
+| **mutation capability proof** | Required permissions for the target role present; forbidden ops still forbidden |
+| **rollback instructions** | How to restore previous SHA and re-verify health |
+
+Suggested durable marker:
+
+```text
+## MCP STABLE RUNTIME PROMOTION (#615)
+
+Status: COMPLETED | ROLLED_BACK | ABORTED
+Previous-SHA: <full sha>
+Promoted-SHA: <full sha>
+Source-PR: <number>
+Source-Branch: <name>
+Reload-Method: <text>
+Health-Proof: client_namespace whoami OK / assess_mcp_namespace_health OK
+Identity-Proof: profile=<name> user=<name>
+Workspace-Proof: root=<path>
+Mutation-Proof: allowed_ops include <…>; forbidden include <…>
+Rollback: checkout <previous sha>; reload method <…>; re-run health/identity proofs
+Operator: <username>
+Timestamp: <ISO-8601>
+```
+
+### 2.5 Unhealthy stable runtime → stop work
+
+If the stable MCP runtime is **unhealthy**, including any of:
+
+- client-namespace probes fail
+- wrong identity / wrong profile
+- wrong workspace root
+- missing mutation capability for the intended role
+- persistent EOF after **client reconnect**
+- **master parity is stale** (`startup_head` behind on-disk `master` / `restart_required` from the parity gate)
+
+then:
+
+1. **Normal PR / review / merge / issue-mutation work must stop immediately.**
+2. The session **reports** the unhealthy/stale state (tool error, CTH, or operator handoff) with startup vs current head when known.
+3. Do **not** improvise: no LLM process kill/restart, no dev-worktree MCP for production mutations, no env escape hatches, no manual gate bypass.
+4. Resume only after:
+   - an **operator** restores the runtime (see §2.6 for routine post-merge parity reload), **or**
+   - a **controlled** promotion/rollback completes with the §2.4 promotion record, **or**
+   - a controller invokes the narrow **bootstrap review path** (#557) when the defect is self-hosted and documented,
+   - **and** the session re-verifies health (and master parity when applicable) before the next mutation.
+
+### 2.6 Routine post-merge master-parity staleness (operator reload, not promotion)
+
+**Symptom:** After merges land on `master`, a long-lived stable MCP process still runs the pre-merge `startup_head`. The master-parity gate marks the server **stale** / `restart_required` and **blocks mutations**.
+
+**Sanctioned response (authoritative):**
+
+| Step | Actor | Action |
+|------|-------|--------|
+| 1 | LLM session | Mutations stop immediately when the gate reports stale. |
+| 2 | LLM session | Report the stale state (startup head, current master head, that operator reload is required). Do not retry mutations. |
+| 3 | **Operator** | Reload/restart the **stable** control MCP so it loads current `master` (supervised client/daemon reload). |
+| 4 | LLM session | Resume only after startup/current-head parity is verified (e.g. `gitea_get_runtime_context` / parity assessment shows in parity). |
+
+**Not a §2.4 promotion:** Catching the already-designated stable control checkout up to a newly advanced `master` is a **routine operator reload** of the stable runtime. It does **not** require the nine-field promotion ledger. Use §2.4 only when **changing which unpromoted/dev revision** becomes the stable control runtime (new source branch/PR into the stable designation).
+
+**Code note:** `master_parity_gate.py` may still say “restart the server” in machine-facing reason strings. That string names the **operator recovery action**, not an LLM self-service instruction. This ADR and the runbooks define the actor split.
+
+## 3. Relationship to other controls
+
+| Doc / mechanism | Interaction |
+|-----------------|-------------|
+| Namespace health (#543) | Proves IDE client can call tools; does not authorize restart |
+| EOF recovery | Reconnect only; no process kill |
+| Daemon import guard (#558) | Mutations require sanctioned daemon; not a bare shell import |
+| Bootstrap path (#557) | Only controller-authorized exception when live runtime cannot review its own fix |
+| Allocator / control-plane ADR | Coordination DB is separate; still depends on a healthy MCP surface for Gitea writes |
+
+## 4. Consequences
+
+### Positive
+
+- Predictable control plane for concurrent LLMs
+- Clear operator-only promotion gate with rollback
+- Aligns session behavior with health/EOF docs already landed
+
+### Costs
+
+- LLM sessions must wait when runtime is sick (no DIY restart)
+- Operators must maintain promotion discipline and dual-runtime config if they use a dev MCP
+
+### Non-goals
+
+- Does not ban operator-supervised restarts during incidents
+- Does not replace CI or code review for MCP changes
+- Does not authorize editing stable checkout “because tests need a quick fix”
+
+## 5. Implementation follow-ups (optional tooling)
+
+These may land in later issues; the **policy binds sessions now**:
+
+1. Session preflight that refuses mutations if workspace root equals a `branches/` feature worktree configured as “dev only.”
+2. Explicit `runtime_kind=stable|dev` in MCP config and `gitea_whoami` profile metadata.
+3. Promotion checklist script that emits the durable promotion marker fields.
+
+**Not optional (issue #615 acceptance criterion 2):** operator guide and runbooks **must** cross-link this ADR (see §6). Cross-links are documentation acceptance, not deferred tooling.
+
+## 6. Acceptance for this ADR
+
+1. Document merged under `docs/architecture/mcp-stable-control-runtime-policy-adr.md`.
+2. **Operator guide / runbooks cross-link this ADR** (`docs/wiki/Operator-Guide.md`, `docs/wiki/Runbooks.md`, `docs/llm-workflow-runbooks.md`).
+3. Issue #615 references this path.
+4. LLM/operator runbooks treat kill/restart/relaunch-from-worktree as **LLM violations**; process restart is **operator-owned**.
+5. Unhealthy runtime (including **stale master parity**) stops normal mutation work until operator restore/reload, promotion/rollback, or #557 bootstrap — then re-verify parity before mutating.
+6. Routine post-merge parity reload is documented as operator reload (§2.6), not an LLM self-restart and not a full §2.4 promotion.
+
+## 7. Document history
+
+| Date | Change |
+|------|--------|
+| 2026-07-09 | Initial ADR: stable vs dev runtime, forbidden session actions, promotion proof fields, stop-work rule |
+| 2026-07-16 | Review 443 remediation (#615 / PR #616): mandatory cross-links; LLM vs operator restart split; routine post-merge parity staleness (§2.6); stale parity in §2.5 unhealthy triggers |

docs/llm-workflow-runbooks.md

@@ -195,7 +195,7 @@ To avoid the bottleneck of relaunching/restarting the MCP server to switch betwe
   `gitea_reconcile_already_landed_pr` after ancestry proof — not for normal
   review or author workflows.
 
-* **Fallback:** If the dual-profile MCP launcher pattern is not supported or configured in the client, the LLM must relaunch or restart the client/MCP with the correct profile environment variable before claiming or working on any tasks.
+* **Fallback (operator-owned):** If the dual-profile MCP launcher pattern is not supported or configured in the client, **do not** have the LLM relaunch or restart the client/MCP. The LLM **stops** role-switching work, reports that the correct static namespace is missing, and waits for an **operator** to configure dual namespaces or reload the client with the correct `GITEA_MCP_PROFILE` for that role. Process restart/relaunch is operator-owned under the [stable control runtime ADR](architecture/mcp-stable-control-runtime-policy-adr.md) (#615).
 
 ## Setup runbook — interactive menu
 
@@ -1200,10 +1200,13 @@ When a mutation blocks on workspace binding:
 
 1. Read the error — it names the **resolved workspace path**, **role
    namespace**, and **binding source** (tool arg, env var, or process root).
-2. Reconnect or relaunch the correct namespace MCP server from the intended
-   workspace (or set the role-specific env var before launch).
-3. Pass `worktree_path` on reviewer/merger mutation tools when the active
-   branches/ worktree differs from the MCP process root.
+2. **LLM-allowed:** pass `worktree_path` on reviewer/merger mutation tools when
+   the active `branches/` worktree differs from the MCP process root; **client
+   reconnect** after transport EOF only (no process kill).
+3. **Operator-owned:** if the wrong namespace process was launched, or a role-
+   specific `GITEA_*_WORKTREE` must be set at process start, an **operator**
+   reloads/relaunches the correct static namespace MCP. LLM sessions must not
+   kill or restart MCP processes (see [stable control runtime ADR](architecture/mcp-stable-control-runtime-policy-adr.md)).
 4. **Do not** clean, reset, or discard foreign role worktrees to unblock your
    own namespace — that destroys another agent's WIP.
 
@@ -1213,9 +1216,10 @@ When posting a Canonical Thread Handoff after a binding blocker:
 
 - State which namespace was active (author / reviewer / merger / reconciler).
 - Quote the resolved workspace path and binding source from the error.
-- Name the safe reconnect action (relaunch MCP from `branches/...`, set
-  `GITEA_*_WORKTREE`, or pass `worktree_path`).
-- Explicitly note that foreign worktrees must not be cleaned to unblock.
+- Name the safe next action: pass `worktree_path` if that unblocks the tool, or
+  request an **operator** reload of the correct namespace MCP / env binding.
+- Explicitly note that foreign worktrees must not be cleaned to unblock, and
+  that the LLM must not self-restart the MCP process.
 
 ## Safety notes
 
@@ -1226,6 +1230,7 @@ When posting a Canonical Thread Handoff after a binding blocker:
 
 ## Related documents
 
+- [`architecture/mcp-stable-control-runtime-policy-adr.md`](architecture/mcp-stable-control-runtime-policy-adr.md) — stable control runtime vs dev runtime; LLM must not kill/restart MCP; operator-owned reload and promotions; routine post-merge parity staleness (#615).
 - [`reviewer-handoff-consistency.md`](reviewer-handoff-consistency.md) — reject contradictory reviewer handoffs (#501).
 - [`issue-acceptance-gate.md`](issue-acceptance-gate.md) — controller issue-acceptance audit after PR merge (#500).
 - [`../skills/llm-project-workflow/SKILL.md`](../skills/llm-project-workflow/SKILL.md) — portable cross-project LLM workflow skill.

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.

docs/wiki/Operator-Guide.md

@@ -14,6 +14,7 @@ Handbook for LLM operators and human developers using the Gitea-Tools MCP server
 4. **No self-review / no self-merge** — The authenticated Gitea user must not approve or merge a PR they authored.
 5. **Follow the gates** — Prompts express intent; MCP tools enforce safety. Never bypass gates via prompt instructions.
 6. **Global LLM Worktree Rule** — Main checkout stays on `master`/`main`/`dev`; all mutations happen under `branches/`. Prove project root, `cwd`, branch, stable main-checkout branch, and session worktree path before editing. No exceptions.
+7. **Stable control runtime** — Real Gitea mutations use only the **stable** MCP control runtime. LLM sessions must not kill, restart, or relaunch MCP processes, edit the stable checkout, or use a dev MCP for production mutations. See the policy ADR: [mcp-stable-control-runtime-policy-adr.md](../architecture/mcp-stable-control-runtime-policy-adr.md) (#615).
 
 ## Supported Gitea instances
 
@@ -29,4 +30,5 @@ Always pass `remote` explicitly on tool calls. The server default is `dadeschool
 1. `gitea_whoami` — confirm authenticated user and profile.
 2. `gitea_get_runtime_context` — allowed/forbidden operations for this session.
 3. `gitea_resolve_task_capability` — prove the session may perform the planned task.
-4. For reviewer work: dry-run validation (`gitea_dry_run_pr_review`) before live review mutations.
+4. For reviewer work: dry-run validation (`gitea_dry_run_pr_review`) before live review mutations.
+5. Confirm master parity / runtime health when tools report stale or unhealthy control runtime — stop mutations and request an **operator** reload of the stable MCP (see [stable control runtime ADR](../architecture/mcp-stable-control-runtime-policy-adr.md)).

docs/wiki/Runbooks.md

@@ -12,6 +12,17 @@
 4. `gitea_mark_final_review_decision` → approve via `gitea_review_pr`.
 5. `gitea_merge_pr` with pinned head SHA and `confirmation="MERGE PR <n>"`.
 
+## Stable MCP control runtime (#615)
+
+Policy ADR: [mcp-stable-control-runtime-policy-adr.md](../architecture/mcp-stable-control-runtime-policy-adr.md).
+
+| Situation | Who acts | What to do |
+|-----------|----------|------------|
+| Transport EOF / missing tools | LLM | **Client reconnect only** — do not kill PIDs |
+| Wrong profile / dual-namespace missing | Operator | Configure or reload the correct static namespace(s) |
+| Master parity stale after merge | LLM stops + reports; **operator** reloads stable MCP | Resume only after parity re-verified |
+| Promote unpromoted MCP code to stable | Operator / release-manager only | Full §2.4 promotion record |
+
 ## Gitea Wiki sync
 
 The Gitea Wiki mirrors `docs/wiki/` (source of truth). After merging wiki changes:

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]:

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,

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

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,6 +60,15 @@ 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",
@@ -68,10 +77,22 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
         "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": {
@@ -84,7 +105,7 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
     },
     "adopt_merger_pr_lease": {
         "permission": "gitea.pr.comment",
-        "role": "reviewer",
+        "role": "merger",
     },
     # #691: guarded non-owner cleanup of obsolete comment-backed reviewer leases.
     # Apply path posts lease release + audit comments (gitea.pr.comment).
@@ -127,6 +148,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": {

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_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_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_stable_control_runtime_policy_docs.py

@@ -0,0 +1,139 @@
+"""Documentation acceptance for the stable control runtime ADR (#615 / PR #616).
+
+Enforces review 443 remediation:
+
+* F1 — operator guide / runbooks cross-link the ADR (issue #615 AC2).
+* F2 — LLM sessions are not instructed to kill/restart/relaunch MCP; process
+  restart is operator-owned; ADR is the authoritative split.
+* F3 — routine post-merge master-parity staleness has a sanctioned response
+  (stop → report → operator reload → re-verify parity) and is not a full
+  promotion ledger requirement.
+"""
+from pathlib import Path
+
+REPO_ROOT = Path(__file__).resolve().parent.parent
+ADR = (
+    REPO_ROOT
+    / "docs"
+    / "architecture"
+    / "mcp-stable-control-runtime-policy-adr.md"
+)
+ADR_REL = "architecture/mcp-stable-control-runtime-policy-adr.md"
+ADR_BASENAME = "mcp-stable-control-runtime-policy-adr.md"
+
+CROSS_LINK_DOCS = (
+    REPO_ROOT / "docs" / "wiki" / "Operator-Guide.md",
+    REPO_ROOT / "docs" / "wiki" / "Runbooks.md",
+    REPO_ROOT / "docs" / "llm-workflow-runbooks.md",
+)
+
+
+def _read(path: Path) -> str:
+    assert path.is_file(), f"missing {path.relative_to(REPO_ROOT)}"
+    return path.read_text(encoding="utf-8")
+
+
+def test_adr_exists_with_policy_core():
+    text = _read(ADR)
+    assert text.lstrip().startswith("#"), "ADR lacks a title"
+    assert "#615" in text
+    assert "stable control runtime" in text.lower()
+    assert "2.3" in text and "2.4" in text and "2.5" in text and "2.6" in text
+
+
+def test_f1_operator_docs_cross_link_adr():
+    for path in CROSS_LINK_DOCS:
+        text = _read(path)
+        assert ADR_BASENAME in text, (
+            f"{path.relative_to(REPO_ROOT)} must cross-link {ADR_BASENAME} "
+            f"(issue #615 acceptance criterion 2)"
+        )
+
+
+def test_f1_adr_lists_cross_links_as_acceptance_not_optional_tooling():
+    text = _read(ADR)
+    # Acceptance section must require guide/runbook cross-links.
+    assert "Operator guide / runbooks cross-link" in text or (
+        "operator guide" in text.lower() and "cross-link" in text.lower()
+        and "Acceptance" in text
+    )
+    # Cross-link must not remain only as optional tooling item #4.
+    optional = text.split("## 5. Implementation follow-ups", 1)[-1].split(
+        "## 6. Acceptance", 1
+    )[0]
+    assert "Operator Guide wiki cross-link to this ADR" not in optional, (
+        "ADR §5 must not list operator-guide cross-link as optional tooling"
+    )
+    assert "Not optional" in text or "must** cross-link" in text.lower() or (
+        "must cross-link" in text.lower()
+    )
+
+
+def test_f2_runbook_fallback_is_operator_owned_not_llm_restart():
+    runbooks = _read(REPO_ROOT / "docs" / "llm-workflow-runbooks.md")
+    # Forbidden historical self-service instruction.
+    forbidden = (
+        "the LLM must relaunch or restart the client/MCP with the correct "
+        "profile environment variable before claiming or working on any tasks"
+    )
+    assert forbidden not in runbooks, (
+        "llm-workflow-runbooks must not instruct the LLM to relaunch/restart MCP"
+    )
+    assert "operator-owned" in runbooks.lower() or "Operator-owned" in runbooks
+    assert ADR_BASENAME in runbooks
+
+
+def test_f2_workspace_rebind_does_not_require_llm_process_relaunch():
+    runbooks = _read(REPO_ROOT / "docs" / "llm-workflow-runbooks.md")
+    section = runbooks.split("### Safe reconnect / rebind procedure", 1)[-1]
+    section = section.split("## Safety notes", 1)[0]
+    # Must not tell the LLM alone to relaunch the MCP process as step 2.
+    assert "Reconnect or relaunch the correct namespace MCP server from the intended" not in section
+    assert "Operator-owned" in section or "operator" in section.lower()
+    assert "worktree_path" in section
+    collapsed = " ".join(section.lower().split())
+    assert "client reconnect" in collapsed
+
+
+def test_f2_adr_forbids_llm_restart_and_supersedes_self_service_relaunch():
+    text = _read(ADR)
+    lower = text.lower()
+    assert "must not" in lower and "restart" in lower
+    assert "operator" in lower and "reload" in lower
+    assert "supersedes" in lower
+    assert "llm" in lower
+
+
+def test_f3_adr_defines_post_merge_parity_staleness_response():
+    text = _read(ADR)
+    lower = text.lower()
+    assert "2.6" in text
+    assert "post-merge" in lower or "post merge" in lower
+    assert "parity" in lower and "stale" in lower
+    # Sanctioned steps: stop, report, operator reload, re-verify.
+    assert "stop" in lower
+    assert "report" in lower
+    assert "operator" in lower
+    assert "parity is verified" in lower or "re-verif" in lower or (
+        "startup/current-head" in lower
+    )
+    # Not a full promotion ledger for routine reload.
+    assert "not a §2.4 promotion" in lower or "not a section 2.4 promotion" in lower or (
+        "not a §2.4" in text or "Not a §2.4 promotion" in text
+    )
+    # Stale parity listed among unhealthy triggers.
+    assert "master parity is stale" in lower or "stale master parity" in lower
+
+
+def test_f3_adr_forbids_llm_bypass_of_parity_gate():
+    text = _read(ADR)
+    lower = text.lower()
+    assert "bypass" in lower or "self-reset" in lower or "self-service" in lower
+    assert "parity" in lower
+
+
+def test_cross_links_do_not_embed_secrets_or_raw_hosts_in_wiki_snippets():
+    for path in CROSS_LINK_DOCS:
+        text = _read(path)
+        for marker in ("ghp_", "BEGIN PRIVATE KEY", "Authorization: Bearer"):
+            assert marker not in text, f"{path} contains {marker!r}"

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

@@ -0,0 +1,205 @@
+"""Invariant tests pinning task_capability_map role assignments (#722/#723).
+
+Added with the operator-authorized break-glass repair for incident #722:
+commit 970e68b remapped ten reviewer tasks to ``role="merger"`` while every
+configured merger profile forbids the review permissions, so no configured
+profile could resolve any formal review task (``matching_configured_profile``
+was empty repository-wide). These tests fail loudly if that class of
+regression recurs:
+
+- every role-exclusive formal-review task must be satisfiable by at least one
+  canonical role profile (permission AND role together);
+- the capability map must agree with ``role_session_router`` task sets;
+- ``adopt_merger_pr_lease`` stays merger-only (the legitimate hunk of
+  970e68b, preserved by the repair);
+- merger profiles must not be able to resolve review_pr/approve_pr.
+"""
+
+import unittest
+
+import gitea_config
+from role_session_router import MERGER_TASKS, REVIEWER_TASKS
+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):
+# reviewers review/approve/request changes but never merge; mergers merge but
+# never review/approve/request changes.
+CANONICAL_ROLE_PROFILES = {
+    "author": {
+        "allowed": [
+            "gitea.read",
+            "gitea.branch.create",
+            "gitea.branch.push",
+            "gitea.repo.commit",
+            "gitea.pr.create",
+            "gitea.pr.comment",
+            "gitea.issue.create",
+            "gitea.issue.comment",
+            "gitea.issue.close",
+        ],
+        "forbidden": [
+            "gitea.pr.approve",
+            "gitea.pr.request_changes",
+            "gitea.pr.merge",
+        ],
+    },
+    "reviewer": {
+        "allowed": [
+            "gitea.read",
+            "gitea.pr.review",
+            "gitea.pr.approve",
+            "gitea.pr.request_changes",
+            "gitea.pr.comment",
+            "gitea.issue.comment",
+        ],
+        "forbidden": [
+            "gitea.branch.create",
+            "gitea.branch.push",
+            "gitea.repo.commit",
+            "gitea.pr.create",
+            "gitea.pr.merge",
+        ],
+    },
+    "merger": {
+        "allowed": [
+            "gitea.read",
+            "gitea.pr.merge",
+            "gitea.pr.comment",
+            "gitea.issue.comment",
+        ],
+        "forbidden": [
+            "gitea.branch.create",
+            "gitea.branch.push",
+            "gitea.repo.commit",
+            "gitea.pr.create",
+            "gitea.pr.approve",
+            "gitea.pr.review",
+            "gitea.pr.request_changes",
+        ],
+    },
+    "reconciler": {
+        "allowed": [
+            "gitea.read",
+            "gitea.pr.close",
+            "gitea.pr.comment",
+            "gitea.issue.comment",
+            "gitea.branch.delete",
+            "gitea.decision_lock.irrecoverable_recovery",
+        ],
+        "forbidden": [
+            "gitea.pr.approve",
+            "gitea.pr.merge",
+            "gitea.pr.review",
+            "gitea.pr.request_changes",
+            "gitea.pr.create",
+            "gitea.branch.create",
+            "gitea.branch.push",
+            "gitea.repo.commit",
+        ],
+    },
+}
+
+# Role-exclusive formal-review tasks (mirrors the resolver's role-exclusive
+# handling for review work): permission alone is not enough — the profile's
+# role kind must also match, so both dimensions are pinned here.
+FORMAL_REVIEW_TASKS = (
+    "review_pr",
+    "approve_pr",
+    "request_changes_pr",
+    "blind_pr_queue_review",
+    "pr_queue_cleanup",
+    "pr-queue-cleanup",
+)
+
+
+def _profile_satisfies(role_name, task):
+    """True when the canonical *role_name* profile can perform *task*."""
+    profile = CANONICAL_ROLE_PROFILES[role_name]
+    ok, _reason = gitea_config.check_operation(
+        required_permission(task), profile["allowed"], profile["forbidden"]
+    )
+    return ok and role_name == required_role(task)
+
+
+class TestFormalReviewProfileCoverage(unittest.TestCase):
+    """#722: some configured profile must be able to formally review."""
+
+    def test_every_formal_review_task_has_a_satisfying_role_profile(self):
+        for task in FORMAL_REVIEW_TASKS:
+            with self.subTest(task=task):
+                satisfying = [
+                    role
+                    for role in CANONICAL_ROLE_PROFILES
+                    if _profile_satisfies(role, task)
+                ]
+                self.assertTrue(
+                    satisfying,
+                    f"no canonical role profile satisfies both permission "
+                    f"{required_permission(task)!r} and role "
+                    f"{required_role(task)!r} for task {task!r} — formal "
+                    f"review would be impossible for every configured "
+                    f"profile (incident #722)",
+                )
+
+    def test_formal_review_tasks_are_reviewer_role(self):
+        for task in FORMAL_REVIEW_TASKS:
+            with self.subTest(task=task):
+                self.assertEqual(required_role(task), "reviewer")
+
+
+class TestMapRouterAgreement(unittest.TestCase):
+    """#723 AC2: the map and the role session router must not drift."""
+
+    def test_reviewer_tasks_map_to_reviewer_role(self):
+        for task in sorted(REVIEWER_TASKS):
+            with self.subTest(task=task):
+                self.assertEqual(
+                    required_role(task),
+                    "reviewer",
+                    f"router classifies {task!r} as a reviewer task but the "
+                    f"capability map assigns role {required_role(task)!r}",
+                )
+
+    def test_merger_tasks_map_to_merger_role(self):
+        for task in sorted(MERGER_TASKS):
+            with self.subTest(task=task):
+                self.assertEqual(
+                    required_role(task),
+                    "merger",
+                    f"router classifies {task!r} as a merger task but the "
+                    f"capability map assigns role {required_role(task)!r}",
+                )
+
+
+class TestMergerBoundary(unittest.TestCase):
+    """Preserve the legitimate hunk of 970e68b and the merger fence."""
+
+    def test_adopt_merger_pr_lease_requires_merger_role(self):
+        self.assertEqual(required_role("adopt_merger_pr_lease"), "merger")
+        self.assertEqual(
+            required_permission("adopt_merger_pr_lease"), "gitea.pr.comment"
+        )
+
+    def test_merge_pr_requires_merger_role(self):
+        self.assertEqual(required_role("merge_pr"), "merger")
+        self.assertEqual(required_permission("merge_pr"), "gitea.pr.merge")
+
+    def test_merger_profile_cannot_resolve_formal_review_tasks(self):
+        merger = CANONICAL_ROLE_PROFILES["merger"]
+        for task in ("review_pr", "approve_pr", "request_changes_pr"):
+            with self.subTest(task=task):
+                ok, reason = gitea_config.check_operation(
+                    required_permission(task),
+                    merger["allowed"],
+                    merger["forbidden"],
+                )
+                self.assertFalse(
+                    ok,
+                    f"merger profile must not hold {task!r} permission "
+                    f"(got reason {reason!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 {