fix(cleanup): harden readback scope and ownership fail-closed (#687)

Require branch-scoped post-delete not-found (not generic/repo/host 404),
emit consistent top-level cleanup fields, fail closed on ownership inventory
errors, never auto-reclaim expired control-plane leases, include active
comment-backed reviewer leases, apply the same gates to reconcile_merged_cleanups,
and match ownership with normalized host identity.

branch_cleanup_guard.py

@@ -7,6 +7,19 @@ from typing import Any
 
 PROTECTED_BRANCHES = frozenset({"master", "main", "dev"})
 
+# Evidence / preservation branches must never be removed by cleanup tools
+# (e.g. chore/issue-681-preserve-review-session-wip).
+_PRESERVATION_MARKERS = ("preserve", "preservation", "evidence")
+
+
+def is_preservation_or_evidence_branch(branch: str | None) -> bool:
+    """Return True when *branch* is a preservation/evidence ref that must stay."""
+    if not branch:
+        return False
+    name = str(branch).lower()
+    return any(marker in name for marker in _PRESERVATION_MARKERS)
+
+
 _RAW_BRANCH_DELETE_PATTERNS = (
     re.compile(r"\bgit(?:\s+-C\s+\S+)?\s+branch\s+-[dD]\b[^\n\r]*", re.I),
     re.compile(r"\bgit(?:\s+-C\s+\S+)?\s+push\b[^\n\r]*\s--delete\b[^\n\r]*", re.I),
@@ -72,6 +85,11 @@ def assess_merged_pr_branch_cleanup(
         reasons.append("PR head branch is missing")
     if head_branch in protected:
         reasons.append(f"branch '{head_branch}' is protected")
+    if is_preservation_or_evidence_branch(head_branch):
+        reasons.append(
+            f"branch '{head_branch}' is a preservation/evidence branch and "
+            "cannot be deleted through merged-PR cleanup"
+        )
     if head_branch in open_pr_heads:
         reasons.append("an open PR still references this head branch")
     if head_on_target is False:
@@ -96,3 +114,490 @@ def assess_merged_pr_branch_cleanup(
         "block_reasons": reasons,
         "recommended_action": "delete_remote_branch" if safe else "keep_remote_branch",
     }
+
+
+# ---------------------------------------------------------------------------
+# #687 remediation: post-delete readback + active ownership protection
+# ---------------------------------------------------------------------------
+
+READBACK_NOT_FOUND = "not_found"
+READBACK_EXISTS = "exists"
+READBACK_AUTHENTICATION = "authentication_error"
+READBACK_AUTHORIZATION = "authorization_error"
+READBACK_TRANSPORT = "transport_error"
+READBACK_UNEXPECTED = "unexpected_response"
+READBACK_AMBIGUOUS_404 = "ambiguous_not_found"
+
+# Scope of a 404: only branch-scoped absence may set verified_absent.
+NOT_FOUND_SCOPE_BRANCH = "branch"
+NOT_FOUND_SCOPE_REPOSITORY = "repository"
+NOT_FOUND_SCOPE_HOST = "host"
+NOT_FOUND_SCOPE_UNKNOWN = "unknown"
+
+ERROR_CLASS_AUTHENTICATION = "authentication"
+ERROR_CLASS_AUTHORIZATION = "authorization"
+ERROR_CLASS_TRANSPORT = "transport"
+ERROR_CLASS_UNEXPECTED = "unexpected"
+
+OWNERSHIP_CATEGORY_AUTHOR_SESSION = "author_session"
+OWNERSHIP_CATEGORY_AUTHOR_LEASE = "author_lease"
+OWNERSHIP_CATEGORY_REVIEWER_LEASE = "reviewer_lease"
+OWNERSHIP_CATEGORY_MERGER_LEASE = "merger_lease"
+OWNERSHIP_CATEGORY_CONTROLLER_LEASE = "controller_lease"
+OWNERSHIP_CATEGORY_RECONCILER_LEASE = "reconciler_lease"
+OWNERSHIP_CATEGORY_WORKTREE_BINDING = "worktree_binding"
+OWNERSHIP_CATEGORY_INVENTORY_ERROR = "ownership_inventory_error"
+
+_ROLE_TO_OWNERSHIP_CATEGORY = {
+    "author": OWNERSHIP_CATEGORY_AUTHOR_LEASE,
+    "reviewer": OWNERSHIP_CATEGORY_REVIEWER_LEASE,
+    "merger": OWNERSHIP_CATEGORY_MERGER_LEASE,
+    "controller": OWNERSHIP_CATEGORY_CONTROLLER_LEASE,
+    "reconciler": OWNERSHIP_CATEGORY_RECONCILER_LEASE,
+}
+
+_ACTIVE_OWNERSHIP_STATUSES = frozenset(
+    {"active", "live", "claimed", "in_progress", "working", "pushing", "pushed"}
+)
+_TERMINAL_OWNERSHIP_STATUSES = frozenset(
+    {"released", "abandoned", "done", "blocked", "terminal", "closed"}
+)
+_EXPIRED_STATUSES = frozenset({"expired"})
+_STALE_STATUSES = frozenset({"stale", "stale_dead_process", "stale_missing_worktree"})
+
+
+def _norm_str(value: Any) -> str:
+    return str(value or "").strip()
+
+
+def normalize_host(host: str | None) -> str:
+    """Normalize a host identity for ownership matching (no credentials)."""
+    text = _norm_str(host).lower()
+    for prefix in ("https://", "http://"):
+        if text.startswith(prefix):
+            text = text[len(prefix) :]
+    # Drop path/query if a full URL slipped through.
+    text = text.split("/", 1)[0]
+    text = text.split("?", 1)[0]
+    return text.rstrip(".")
+
+
+def ownership_category_for_role(role: str | None) -> str:
+    """Map a role kind to a non-secret ownership category label."""
+    key = _norm_str(role).lower()
+    return _ROLE_TO_OWNERSHIP_CATEGORY.get(key, f"{key or 'unknown'}_lease")
+
+
+def classify_branch_readback_http_status(
+    status_code: int | None,
+    *,
+    not_found_scope: str | None = None,
+) -> dict[str, Any]:
+    """Classify a GET-branch HTTP status into a secret-free readback result.
+
+    R1: A bare/generic/repository/wrong-host 404 never yields
+    ``verified_absent=True``. Only an authoritative *branch-scoped* not-found
+    (``not_found_scope='branch'``) may verify deletion.
+    """
+    if status_code == 404:
+        scope = _norm_str(not_found_scope).lower() or NOT_FOUND_SCOPE_UNKNOWN
+        if scope == NOT_FOUND_SCOPE_BRANCH:
+            return {
+                "status": READBACK_NOT_FOUND,
+                "error_class": None,
+                "verified_absent": True,
+                "branch_present": False,
+                "not_found_scope": NOT_FOUND_SCOPE_BRANCH,
+                "reasons": [],
+            }
+        # repository / host / unknown / generic 404 — not verified absence
+        reason = {
+            NOT_FOUND_SCOPE_REPOSITORY: (
+                "post-delete readback 404 is repository-scoped, not branch absence"
+            ),
+            NOT_FOUND_SCOPE_HOST: (
+                "post-delete readback 404 is host-scoped, not branch absence"
+            ),
+        }.get(
+            scope,
+            "post-delete readback 404 is ambiguous (not branch-scoped); "
+            "cannot verify absence",
+        )
+        return {
+            "status": READBACK_AMBIGUOUS_404,
+            "error_class": ERROR_CLASS_UNEXPECTED,
+            "verified_absent": False,
+            "branch_present": None,
+            "not_found_scope": scope,
+            "reasons": [reason],
+        }
+    if status_code in (401, 407):
+        return {
+            "status": READBACK_AUTHENTICATION,
+            "error_class": ERROR_CLASS_AUTHENTICATION,
+            "verified_absent": False,
+            "branch_present": None,
+            "reasons": ["post-delete branch readback authentication failed"],
+        }
+    if status_code == 403:
+        return {
+            "status": READBACK_AUTHORIZATION,
+            "error_class": ERROR_CLASS_AUTHORIZATION,
+            "verified_absent": False,
+            "branch_present": None,
+            "reasons": ["post-delete branch readback authorization failed"],
+        }
+    if status_code is not None and 200 <= int(status_code) < 300:
+        return {
+            "status": READBACK_EXISTS,
+            "error_class": None,
+            "verified_absent": False,
+            "branch_present": True,
+            "reasons": ["post-delete readback found branch still present"],
+        }
+    if status_code is not None and int(status_code) >= 500:
+        return {
+            "status": READBACK_TRANSPORT,
+            "error_class": ERROR_CLASS_TRANSPORT,
+            "verified_absent": False,
+            "branch_present": None,
+            "reasons": ["post-delete branch readback transport/upstream failure"],
+        }
+    return {
+        "status": READBACK_UNEXPECTED,
+        "error_class": ERROR_CLASS_UNEXPECTED,
+        "verified_absent": False,
+        "branch_present": None,
+        "reasons": ["post-delete branch readback returned an unexpected response"],
+        "http_status": status_code,
+    }
+
+
+def _extract_http_status(exc: BaseException) -> int | None:
+    """Extract an HTTP status code from an exception chain (secret-free)."""
+    seen: set[int] = set()
+    current: BaseException | None = exc
+    while current is not None and id(current) not in seen:
+        seen.add(id(current))
+        for attr in ("code", "status", "status_code"):
+            value = getattr(current, attr, None)
+            if isinstance(value, int) and 100 <= value <= 599:
+                return value
+        text = str(current) if current is not None else ""
+        match = re.match(r"HTTP\s+(\d{3})\b", text)
+        if match:
+            return int(match.group(1))
+        current = current.__cause__ or current.__context__
+    return None
+
+
+def classify_branch_readback_exception(
+    exc: BaseException,
+    *,
+    not_found_scope: str | None = None,
+) -> dict[str, Any]:
+    """Classify a GET-branch exception without leaking credentials or bodies.
+
+    R1: substring ``404`` / ``not found`` alone never becomes verified_absent.
+    Callers must pass ``not_found_scope='branch'`` only after authoritative
+    proof that the repository/host is still reachable and the 404 is branch-level.
+    """
+    status_code = _extract_http_status(exc)
+    if status_code is None:
+        lower = str(exc).lower() if exc is not None else ""
+        if any(
+            token in lower
+            for token in (
+                "timed out",
+                "timeout",
+                "connection",
+                "network",
+                "temporarily unavailable",
+                "name or service not known",
+                "nodename nor servname",
+            )
+        ):
+            return {
+                "status": READBACK_TRANSPORT,
+                "error_class": ERROR_CLASS_TRANSPORT,
+                "verified_absent": False,
+                "branch_present": None,
+                "reasons": [
+                    "post-delete branch readback transport/upstream failure"
+                ],
+            }
+        if "unauthorized" in lower:
+            status_code = 401
+        elif "forbidden" in lower:
+            status_code = 403
+        elif "404" in lower or "not found" in lower:
+            # Ambiguous: do NOT treat as branch absence without scope proof.
+            status_code = 404
+            if not_found_scope is None:
+                not_found_scope = NOT_FOUND_SCOPE_UNKNOWN
+
+    if status_code is not None:
+        return classify_branch_readback_http_status(
+            status_code, not_found_scope=not_found_scope
+        )
+
+    return {
+        "status": READBACK_UNEXPECTED,
+        "error_class": ERROR_CLASS_UNEXPECTED,
+        "verified_absent": False,
+        "branch_present": None,
+        "reasons": ["post-delete branch readback returned an unexpected response"],
+    }
+
+
+def assess_post_delete_readback(readback: dict[str, Any] | None) -> dict[str, Any]:
+    """Decide whether DELETE success may be reported after branch readback.
+
+    Success requires authoritative branch-scoped not-found
+    (``verified_absent=True``). DELETE HTTP success alone is never enough.
+    Generic/repository/host 404 cannot verify absence (R1).
+    """
+    payload = dict(readback or {})
+    status = _norm_str(payload.get("status")) or READBACK_UNEXPECTED
+    # Only explicit verified_absent flag counts — never infer from status alone
+    # when status is a bare not_found without branch scope proof.
+    verified = bool(payload.get("verified_absent")) is True
+    if verified and status == READBACK_NOT_FOUND:
+        return {
+            "ok": True,
+            "success": True,
+            "verified_absent": True,
+            "readback": {
+                "status": READBACK_NOT_FOUND,
+                "verified_absent": True,
+                "branch_present": False,
+                "error_class": None,
+                "not_found_scope": NOT_FOUND_SCOPE_BRANCH,
+            },
+            "reasons": [],
+        }
+
+    reasons = list(payload.get("reasons") or [])
+    if not reasons:
+        if status == READBACK_EXISTS:
+            reasons = ["post-delete readback found branch still present"]
+        elif status == READBACK_AUTHENTICATION:
+            reasons = ["post-delete branch readback authentication failed"]
+        elif status == READBACK_AUTHORIZATION:
+            reasons = ["post-delete branch readback authorization failed"]
+        elif status == READBACK_TRANSPORT:
+            reasons = ["post-delete branch readback transport/upstream failure"]
+        elif status == READBACK_AMBIGUOUS_404:
+            reasons = [
+                "post-delete readback 404 is not branch-scoped; "
+                "cannot verify absence"
+            ]
+        else:
+            reasons = ["post-delete branch readback could not verify deletion"]
+
+    return {
+        "ok": False,
+        "success": False,
+        "verified_absent": False,
+        "readback": {
+            "status": status,
+            "verified_absent": False,
+            "branch_present": payload.get("branch_present"),
+            "error_class": payload.get("error_class"),
+            "not_found_scope": payload.get("not_found_scope"),
+        },
+        "reasons": reasons,
+        "blocker_kind": "post_delete_readback_failed",
+    }
+
+
+def cleanup_result_envelope(
+    *,
+    success: bool,
+    performed: bool,
+    delete_acknowledged: bool,
+    verified_absent: bool,
+    **extra: Any,
+) -> dict[str, Any]:
+    """R2: consistent top-level cleanup result fields on every return path."""
+    out: dict[str, Any] = {
+        "success": bool(success),
+        "performed": bool(performed),
+        "delete_acknowledged": bool(delete_acknowledged),
+        "verified_absent": bool(verified_absent),
+    }
+    out.update(extra)
+    return out
+
+
+def _repo_matches(
+    record: dict[str, Any],
+    *,
+    remote: str,
+    org: str,
+    repo: str,
+    host: str | None = None,
+) -> bool:
+    """Match ownership record to target remote/org/repo and normalized host."""
+    if (
+        _norm_str(record.get("remote")).lower() != _norm_str(remote).lower()
+        or _norm_str(record.get("org")).lower() != _norm_str(org).lower()
+        or _norm_str(record.get("repo")).lower() != _norm_str(repo).lower()
+    ):
+        return False
+    expected_host = normalize_host(host)
+    record_host = normalize_host(record.get("host") or record.get("host_name"))
+    # When both sides declare a host, they must agree after normalization.
+    # A record host that disagrees with the expected host is out of scope.
+    # Legacy records without host still match when remote/org/repo agree.
+    if expected_host and record_host and expected_host != record_host:
+        return False
+    return True
+
+
+def _branch_matches(record: dict[str, Any], branch: str) -> bool:
+    return _norm_str(record.get("branch")) == _norm_str(branch)
+
+
+def assess_ownership_record_activity(record: dict[str, Any]) -> dict[str, Any]:
+    """Classify one ownership record as blocking or non-blocking.
+
+    Distinguishes active ownership from expired/released/terminal/stale records.
+    Expired/stale records block unless reclaim_allowed is *explicitly* True
+    (O2: never treat missing/unknown reclaim as auto-allowed).
+    """
+    status = _norm_str(record.get("status")).lower()
+    category = _norm_str(record.get("category")) or "unknown"
+    reclaim_allowed = record.get("reclaim_allowed")
+
+    if category == OWNERSHIP_CATEGORY_INVENTORY_ERROR:
+        return {
+            "blocks": True,
+            "status": status or "unknown",
+            "category": category,
+            "reason": "ownership inventory failed closed",
+        }
+
+    if status in _TERMINAL_OWNERSHIP_STATUSES:
+        return {
+            "blocks": False,
+            "status": status,
+            "category": category,
+            "reason": f"{category} ownership is terminal/released ({status})",
+        }
+    if status in _ACTIVE_OWNERSHIP_STATUSES:
+        return {
+            "blocks": True,
+            "status": status,
+            "category": category,
+            "reason": f"active {category} ownership still uses the target branch",
+        }
+    if status in _EXPIRED_STATUSES or status in _STALE_STATUSES:
+        # O2: only explicit reclaim_allowed=True skips the block.
+        if reclaim_allowed is True:
+            return {
+                "blocks": False,
+                "status": status,
+                "category": category,
+                "reason": (
+                    f"{category} ownership is {status} and reclaimable; "
+                    "does not block deletion"
+                ),
+            }
+        return {
+            "blocks": True,
+            "status": status,
+            "category": category,
+            "reason": (
+                f"{status} {category} ownership still protects the target "
+                "branch (reclaim not proven; fail closed)"
+            ),
+        }
+    # Unknown status → fail closed
+    return {
+        "blocks": True,
+        "status": status or "unknown",
+        "category": category,
+        "reason": (
+            f"unclassified {category} ownership status "
+            f"'{status or 'unknown'}'; fail closed"
+        ),
+    }
+
+
+def assess_active_branch_ownership(
+    *,
+    remote: str,
+    org: str,
+    repo: str,
+    branch: str,
+    host: str | None = None,
+    records: list[dict[str, Any]] | None = None,
+) -> dict[str, Any]:
+    """Assess whether any active ownership still uses *branch* in *repo*.
+
+    Matching requires remote/org/repo/branch and, when provided, normalized
+    host identity. Other repositories, hosts, or branches never false-block.
+    """
+    target_branch = _norm_str(branch)
+    expected_host = normalize_host(host)
+    considered: list[dict[str, Any]] = []
+    blocking: list[dict[str, Any]] = []
+    ignored: list[dict[str, Any]] = []
+
+    for raw in records or []:
+        if not isinstance(raw, dict):
+            continue
+        if not _repo_matches(
+            raw, remote=remote, org=org, repo=repo, host=expected_host or None
+        ):
+            ignored.append(
+                {
+                    "category": _norm_str(raw.get("category")) or "unknown",
+                    "reason": "different repository or host scope",
+                }
+            )
+            continue
+        if not _branch_matches(raw, target_branch):
+            ignored.append(
+                {
+                    "category": _norm_str(raw.get("category")) or "unknown",
+                    "reason": "different branch",
+                }
+            )
+            continue
+        activity = assess_ownership_record_activity(raw)
+        entry = {
+            "category": activity["category"],
+            "status": activity["status"],
+            "blocks": activity["blocks"],
+            "reason": activity["reason"],
+        }
+        considered.append(entry)
+        if activity["blocks"]:
+            blocking.append(entry)
+
+    block = bool(blocking)
+    categories = sorted({b["category"] for b in blocking})
+    reasons = [
+        (
+            "active ownership protects branch "
+            f"'{target_branch}': " + "; ".join(b["reason"] for b in blocking)
+        )
+    ] if block else []
+    return {
+        "block": block,
+        "safe_to_delete": not block,
+        "remote": remote,
+        "org": org,
+        "repo": repo,
+        "host": expected_host or None,
+        "branch": target_branch,
+        "blocking_categories": categories,
+        "blocking": blocking,
+        "considered": considered,
+        "ignored_out_of_scope": ignored,
+        "reasons": reasons,
+        "blocker_kind": "active_branch_ownership" if block else None,
+        "recommended_action": "keep_remote_branch" if block else "delete_remote_branch",
+    }

docs/gitea-execution-profiles.md

@@ -238,11 +238,43 @@ narrow operation set:
 - `gitea.issue.comment`
 - `gitea.issue.close`
 - `gitea.pr.close`
+- `gitea.branch.delete` (merged-branch cleanup only — see below)
 
 Forbidden on reconciler profiles: `gitea.pr.approve`, `gitea.pr.merge`,
 `gitea.pr.review`, `gitea.pr.create`, `gitea.branch.push`, and
 `gitea.repo.commit`.
 
+### Merged-branch cleanup ownership (`gitea.branch.delete`)
+
+The reconciler is the repository-supported owner of merged-PR source-branch
+cleanup: `task_capability_map` maps `cleanup_merged_pr_branch` (and
+`reconciliation_cleanup`) to role `reconciler` with permission
+`gitea.branch.delete`. Post-merge branch lifecycle is reconciliation work —
+it happens after the author, reviewer, and merger roles have completed, and
+it must not be reachable from those roles.
+
+Least-privilege constraints:
+
+- `gitea.branch.delete` is granted **only** to reconciler profiles. Author,
+  reviewer, and merger profiles must never hold it; `gitea_delete_branch`
+  and `gitea_cleanup_merged_pr_branch` fail closed on any profile without
+  the permission.
+- Even with the permission, reconciler deletion is only supported through the
+  guarded `gitea_cleanup_merged_pr_branch` path (#514 / #687): the PR must be
+  merged, the head an ancestor of the target, the branch not protected
+  (`master`/`main`/`dev`), the branch not a preservation/evidence ref (e.g.
+  `chore/issue-681-preserve-review-session-wip`), no open PR may still use the
+  head, and an explicit `CLEANUP MERGED PR <n> BRANCH <branch>` confirmation is
+  required. Raw `gitea_delete_branch` is **denied** to reconciler even when
+  `gitea.branch.delete` is present.
+- Raw `git branch -d` / `git push --delete` cleanup remains blocked by
+  `branch_cleanup_guard` and the final-report validator regardless of
+  profile permissions.
+- `gitea.branch.delete` has no short alias in `GITEA_OPERATION_ALIASES`;
+  write it fully qualified in `allowed_operations`. Migration must emit
+  canonical names such as `gitea.pr.close` (never bare `pr.close` /
+  `issue.close`, which the production normalizer rejects or drops).
+
 Launch a static `gitea-reconciler` MCP namespace with
 `GITEA_MCP_PROFILE=prgs-reconciler`. Profile shape is validated by
 `reconciler_profile.assess_reconciler_profile` (#304). Use the
@@ -251,6 +283,159 @@ Launch a static `gitea-reconciler` MCP namespace with
 fresh target-branch fetch, recorded target SHA, and ancestor proof. PRs whose
 heads are not already landed cannot be closed through this path.
 
+### Operational runbook: grant reconciler `gitea.branch.delete` (#687)
+
+Merging a code PR that updates `migrate_profiles.py` / `reconciler_profile.py`
+**does not** change the live operator profile on disk. Apply the profile
+change deliberately, then reconnect the client-managed namespace.
+
+1. **Approved migration / profile-update command** (from the repo root, using
+   the project venv if present):
+
+   ```bash
+   # Dry-run first (default): validates v2 output, writes nothing
+   python3 migrate_profiles.py -i ~/.config/gitea-tools/profiles.json
+
+   # Apply: creates backup then writes migrated v2 config
+   python3 migrate_profiles.py -i ~/.config/gitea-tools/profiles.json -w
+   # Optional explicit paths:
+   # python3 migrate_profiles.py -i ~/.config/gitea-tools/profiles.json \
+   #   -o ~/.config/gitea-tools/profiles.json \
+   #   --backup ~/.config/gitea-tools/profiles.json.bak -w
+   ```
+
+   If the live file is already v2, edit the reconciler identity’s
+   `allowed_operations` / `forbidden_operations` under
+   `environments.<env>.services.gitea.identities.reconciler` (or the
+   `prgs-reconciler` alias target) so allowed includes the canonical set
+   below — then re-validate with a load of the config (see step 3).
+
+2. **Inspect the generated (or edited) profile** — confirm the reconciler
+   identity, for example:
+
+   ```bash
+   python3 - <<'PY'
+   import json
+   from pathlib import Path
+   cfg = json.loads(Path.home().joinpath(".config/gitea-tools/profiles.json").read_text())
+   # v2 environments shape:
+   ident = cfg["environments"]["prgs"]["services"]["gitea"]["identities"]["reconciler"]
+   print("role:", ident.get("role"))
+   print("allowed:", ident.get("allowed_operations"))
+   print("forbidden:", ident.get("forbidden_operations"))
+   PY
+   ```
+
+3. **Validate canonical operation names and least privilege**
+
+   Expected canonical **allowed** (defaults after migration):
+
+   - `gitea.read`
+   - `gitea.pr.close` (required)
+   - `gitea.pr.comment`
+   - `gitea.issue.comment`
+   - `gitea.issue.close`
+   - `gitea.branch.delete` (recommended; cleanup only)
+
+   Expected **forbidden** includes at least: `gitea.pr.approve`,
+   `gitea.pr.merge`, `gitea.pr.review`, `gitea.pr.create`,
+   `gitea.branch.push`, `gitea.repo.commit`.
+
+   No shorthand (`pr.close`, `issue.close`, `pr.comment`) may remain.
+   Validate with the production loader:
+
+   ```bash
+   python3 - <<'PY'
+   import gitea_config, reconciler_profile
+   from pathlib import Path
+   path = str(Path.home() / ".config/gitea-tools/profiles.json")
+   gitea_config.load_config(path)  # fails closed on invalid config
+   # Or assess the reconciler lists directly after extracting them:
+   # print(reconciler_profile.assess_reconciler_profile(allowed, forbidden))
+   PY
+   ```
+
+4. **Merging PR #688 (or any code PR) does not update the live profile.**
+   Code changes only the migration helper, schema, docs, and tests. The
+   operator must still run `migrate_profiles.py -w` or an equivalent
+   authorized edit of `~/.config/gitea-tools/profiles.json`.
+
+5. **Supported apply method:** `python3 migrate_profiles.py … -w` (backup
+   created automatically) **or** operator-authorized edit of the live
+   profiles file after backup. Unsupported: silent mtime tricks, manual
+   process kill to “reload”, or undocumented env overrides.
+
+6. **Backup and validation:** `-w` copies the input to
+   `<input_path>.bak` (or `--backup PATH`) before writing. Re-run
+   `load_config` / `assess_reconciler_profile` after write. Keep the
+   `.bak` until live whoami/capability checks pass.
+
+7. **Client-managed namespace reconnect/reload:** reconnect or reload the
+   IDE MCP client so `gitea-reconciler` restarts from current `master` and
+   the updated `GITEA_MCP_PROFILE=prgs-reconciler` config. Do not hand-launch
+   `mcp_server.py` / `gitea_mcp_server.py` with ad hoc `GITEA_*` env
+   (see #686 / #630).
+
+8. **Live reverification** (through the client-managed `gitea-reconciler`
+   namespace only):
+
+   - `gitea_whoami` → identity + profile `prgs-reconciler`
+   - `gitea_assess_master_parity` → `stale=false`, `restart_required=false`
+   - `gitea_resolve_task_capability(task="cleanup_merged_pr_branch")` →
+     `allowed_in_current_session=true` only when permission and role match
+   - `gitea_resolve_task_capability(task="delete_branch")` →
+     **not** allowed for reconciler (role denial must be enforced)
+
+9. **Guarded cleanup usage** (example for a merged PR whose source branch
+   remains on the remote):
+
+   ```text
+   gitea_cleanup_merged_pr_branch(
+     pr_number=<N>,
+     branch=<exact PR head branch>,
+     confirmation="CLEANUP MERGED PR <N> BRANCH <exact PR head branch>",
+     remote="prgs",
+     org="Scaled-Tech-Consulting",
+     repo="Gitea-Tools",
+     worktree_path="<path under branches/>",
+   )
+   ```
+
+   The tool refuses unmerged PRs, protected branches, preservation/evidence
+   branches, open-PR heads, mismatched branch names, and wrong confirmation.
+
+10. **Prohibitions**
+
+    - No raw `git push --delete`, `git branch -d` / `-D`, or delete refspecs
+    - No arbitrary `gitea_delete_branch` from reconciler
+    - No unsupported profile switching mid-run without full re-preflight
+    - No ad hoc hand-edits of live profiles **unless** operator-authorized,
+      backed up, and revalidated as above
+
+Canonical migrated reconciler example:
+
+```json
+{
+  "role": "reconciler",
+  "allowed_operations": [
+    "gitea.read",
+    "gitea.pr.close",
+    "gitea.pr.comment",
+    "gitea.issue.comment",
+    "gitea.issue.close",
+    "gitea.branch.delete"
+  ],
+  "forbidden_operations": [
+    "gitea.pr.approve",
+    "gitea.pr.merge",
+    "gitea.pr.review",
+    "gitea.pr.create",
+    "gitea.branch.push",
+    "gitea.repo.commit"
+  ]
+}
+```
+
 ## Identity and fail-closed rules
 
 Before **any** mutating action, a workflow must know both:

mcp_native_cleanup_proof.py

@@ -13,7 +13,6 @@ from typing import Any
 
 AUTHORIZED_CLEANUP_TOOLS = frozenset({
     "gitea_cleanup_post_merge_moot_lease",
-    "gitea_cleanup_obsolete_reviewer_comment_lease",
     "gitea_reconcile_merged_cleanups",
     "gitea_delete_branch",
     "gitea_cleanup_merged_pr_branch",

migrate_profiles.py

@@ -20,12 +20,114 @@ if PROJECT_ROOT not in sys.path:
 import gitea_config
 
 
-AUTHOR_DEFAULT_ALLOWED = ["read", "branch", "commit", "push", "open_pr", "comment"]
-AUTHOR_DEFAULT_FORBIDDEN = ["approve", "request_changes", "merge"]
-REVIEWER_DEFAULT_ALLOWED = [
-    "read", "review", "comment", "approve", "request_changes", "merge"
+# Defaults emit *canonical* operation names only. Shorthand that is not in
+# gitea_config.GITEA_OPERATION_ALIASES (e.g. ``pr.close``, ``issue.close``)
+# is silently dropped by the production loader and must never appear here.
+AUTHOR_DEFAULT_ALLOWED = [
+    "gitea.read",
+    "gitea.branch.create",
+    "gitea.repo.commit",
+    "gitea.branch.push",
+    "gitea.pr.create",
+    "gitea.pr.comment",
 ]
-REVIEWER_DEFAULT_FORBIDDEN = ["branch", "commit", "push", "open_pr"]
+AUTHOR_DEFAULT_FORBIDDEN = [
+    "gitea.pr.approve",
+    "gitea.pr.request_changes",
+    "gitea.pr.merge",
+]
+REVIEWER_DEFAULT_ALLOWED = [
+    "gitea.read",
+    "gitea.pr.review",
+    "gitea.pr.comment",
+    "gitea.pr.approve",
+    "gitea.pr.request_changes",
+    "gitea.pr.merge",
+]
+REVIEWER_DEFAULT_FORBIDDEN = [
+    "gitea.branch.create",
+    "gitea.repo.commit",
+    "gitea.branch.push",
+    "gitea.pr.create",
+]
+# Required reconciler ops (read + pr.close) plus recommended comment/close and
+# branch.delete for guarded merged-PR cleanup. All names must normalize via
+# gitea_config.normalize_operation without being dropped.
+RECONCILER_DEFAULT_ALLOWED = [
+    "gitea.read",
+    "gitea.pr.close",
+    "gitea.pr.comment",
+    "gitea.issue.comment",
+    "gitea.issue.close",
+    "gitea.branch.delete",
+]
+RECONCILER_DEFAULT_FORBIDDEN = [
+    "gitea.pr.approve",
+    "gitea.pr.merge",
+    "gitea.pr.review",
+    "gitea.pr.create",
+    "gitea.branch.push",
+    "gitea.repo.commit",
+]
+
+# Migration-only expansions for common shorthands that are *not* in
+# GITEA_OPERATION_ALIASES. Emitted output is always the canonical form so a
+# second canonicalize pass is a no-op (idempotent).
+_MIGRATION_ONLY_ALIASES = {
+    "pr.close": "gitea.pr.close",
+    "pr.comment": "gitea.pr.comment",
+    "issue.close": "gitea.issue.close",
+    "branch.delete": "gitea.branch.delete",
+}
+
+# Reconciler required ops that must survive migration (from reconciler_profile).
+RECONCILER_REQUIRED_CANONICAL = ("gitea.read", "gitea.pr.close")
+
+
+def canonicalize_operation(op: str) -> str:
+    """Return a canonical operation name accepted by the production loader.
+
+    Fail closed on unknown/ambiguous spellings so required permissions cannot
+    be silently dropped by ``check_operation`` later.
+    """
+    if not isinstance(op, str) or not op.strip():
+        raise ValueError("operation must be a non-empty string (fail closed)")
+    op = op.strip()
+    try:
+        return gitea_config.normalize_operation(op)
+    except gitea_config.ConfigError:
+        pass
+    if op in _MIGRATION_ONLY_ALIASES:
+        return _MIGRATION_ONLY_ALIASES[op]
+    raise ValueError(
+        f"operation {op!r} cannot be canonicalized for migration "
+        "(unknown/ambiguous; fail closed — production loader would drop it)"
+    )
+
+
+def canonicalize_operations(ops, *, context: str = "operations") -> list[str]:
+    """Canonicalize a list of operations; preserve order, drop duplicates."""
+    if not isinstance(ops, list):
+        raise ValueError(f"{context} must be a list (fail closed)")
+    out: list[str] = []
+    seen: set[str] = set()
+    for entry in ops:
+        canon = canonicalize_operation(entry)
+        if canon not in seen:
+            seen.add(canon)
+            out.append(canon)
+    return out
+
+
+def _assert_reconciler_required_survive(allowed: list[str], profile_name: str) -> None:
+    """Fail visibly when migration would leave a reconciler without required ops."""
+    missing = [op for op in RECONCILER_REQUIRED_CANONICAL if op not in set(allowed)]
+    if missing:
+        raise ValueError(
+            f"Profile '{profile_name}' (reconciler) is missing required "
+            f"operation(s) after migration: {missing}. Refusing to emit a "
+            "profile that would silently fail pr.close / read (fail closed)."
+        )
 
 
 def infer_role(name, execution_profile):
@@ -90,9 +192,11 @@ def migrate_v1_to_v2(v1_data):
             ident_name = "reviewer"
         elif role == "author":
             ident_name = "author"
+        elif role == "reconciler":
+            ident_name = "reconciler"
         else:
             role = prof.get("role")
-            if role not in (None, "author", "reviewer"):
+            if role not in (None, "author", "reviewer", "reconciler"):
                 raise ValueError(
                     f"Profile '{name}' has unsupported role {role!r}"
                 )
@@ -124,20 +228,35 @@ def migrate_v1_to_v2(v1_data):
                 raise ValueError(
                     f"Profile '{name}' operation fields must be lists"
                 )
-            identity_data["allowed_operations"] = list(allowed)
-            identity_data["forbidden_operations"] = list(forbidden)
+            try:
+                identity_data["allowed_operations"] = canonicalize_operations(
+                    allowed, context=f"profile '{name}' allowed_operations"
+                )
+                identity_data["forbidden_operations"] = canonicalize_operations(
+                    forbidden, context=f"profile '{name}' forbidden_operations"
+                )
+            except ValueError as exc:
+                raise ValueError(f"Profile '{name}': {exc}") from exc
         elif role == "author":
             identity_data["allowed_operations"] = list(AUTHOR_DEFAULT_ALLOWED)
             identity_data["forbidden_operations"] = list(AUTHOR_DEFAULT_FORBIDDEN)
         elif role == "reviewer":
             identity_data["allowed_operations"] = list(REVIEWER_DEFAULT_ALLOWED)
             identity_data["forbidden_operations"] = list(REVIEWER_DEFAULT_FORBIDDEN)
+        elif role == "reconciler":
+            identity_data["allowed_operations"] = list(RECONCILER_DEFAULT_ALLOWED)
+            identity_data["forbidden_operations"] = list(RECONCILER_DEFAULT_FORBIDDEN)
         else:
             raise ValueError(
                 f"Profile '{name}' has no explicit operation lists and no "
                 "unambiguous author/reviewer role marker (fail closed)"
             )
 
+        if role == "reconciler":
+            _assert_reconciler_required_survive(
+                identity_data["allowed_operations"], name
+            )
+
         # Nest inside environments/services structure
         env = environments.setdefault(env_name, {})
         services = env.setdefault("services", {})

reconciler_profile.py

@@ -18,6 +18,11 @@ RECONCILER_RECOMMENDED_OPERATIONS = (
     "gitea.pr.comment",
     "gitea.issue.comment",
     "gitea.issue.close",
+    # Merged-branch cleanup is reconciler-owned (task_capability_map maps
+    # cleanup_merged_pr_branch -> reconciler). The permission is only
+    # exercisable through the guarded gitea_cleanup_merged_pr_branch path
+    # (#514): merged proof, protected-branch refusal, explicit confirmation.
+    "gitea.branch.delete",
 )
 
 RECONCILER_FORBIDDEN_OPERATIONS = (

reviewer_pr_lease.py

@@ -524,26 +524,13 @@ def assess_lease_inventory(
         "reclaimable_review_leases": reclaimable,
     }
 
-# Canonical next-action vocabulary for reviewer lease handoff (#599, #691).
+# Canonical next-action vocabulary for reviewer lease handoff (#599).
 NEXT_ACTION_ACQUIRE = "acquire"
 NEXT_ACTION_WAIT = "wait"
 NEXT_ACTION_RESUME_EXACT_OWNER_SESSION = "resume_exact_owner_session"
 NEXT_ACTION_RELEASE_EXPIRED_LEASE = "release_expired_lease"
 NEXT_ACTION_OPERATOR_AUTHORIZED_CLEANUP = "operator_authorized_cleanup"
 NEXT_ACTION_REPAIR_WORKTREE_BINDING = "repair_worktree_binding"
-NEXT_ACTION_CLEANUP_OBSOLETE_LEASE = "cleanup_obsolete_reviewer_comment_lease"
-
-CLEANUP_OBSOLETE_LEASE_TOOL = "gitea_cleanup_obsolete_reviewer_comment_lease"
-CLEANUP_OBSOLETE_LEASE_CONFIRMATION_PREFIX = "CLEANUP OBSOLETE REVIEWER LEASE "
-
-# Formal terminal review verdicts that complete a leased-head workflow (#691).
-_TERMINAL_REVIEW_VERDICTS = frozenset({
-    "APPROVED",
-    "REQUEST_CHANGES",
-    "approved",
-    "request_changes",
-    "REQUESTCHANGES",
-})
 
 _HANDOFF_CLASSIFICATIONS = frozenset({
     "no_lease",
@@ -552,12 +539,6 @@ _HANDOFF_CLASSIFICATIONS = frozenset({
     "foreign_active",
     "foreign_reclaimable",
     "foreign_expired",
-    "foreign_active_current_head",
-    "foreign_expired_current_head",
-    "foreign_completed_superseded_head",
-    "foreign_expired_superseded_head",
-    "orphaned_owner_missing",
-    "ambiguous_conflicting_evidence",
     "instructed_lease_missing_with_replacement",
     "worktree_binding_mismatch",
 })
@@ -570,476 +551,6 @@ def _norm_path(value: str | None) -> str:
     return os.path.normpath(text.rstrip("/"))
 
 
-def _normalize_verdict(value: str | None) -> str:
-    text = (value or "").strip().upper().replace("-", "_").replace(" ", "_")
-    if text in {"REQUESTCHANGES", "REQUEST_CHANGES", "CHANGES_REQUESTED"}:
-        return "REQUEST_CHANGES"
-    if text in {"APPROVED", "APPROVE"}:
-        return "APPROVED"
-    return text
-
-
-def formal_terminal_review_for_head(
-    formal_reviews: list[dict] | None,
-    *,
-    leased_head: str | None,
-) -> dict[str, Any] | None:
-    """Return the latest undismissed terminal formal review for *leased_head*."""
-    head = _normalize_sha(leased_head)
-    if not head:
-        return None
-    matches: list[dict[str, Any]] = []
-    for review in formal_reviews or []:
-        if review.get("dismissed"):
-            continue
-        verdict = _normalize_verdict(
-            review.get("verdict") or review.get("state") or review.get("body_state")
-        )
-        if verdict not in {"APPROVED", "REQUEST_CHANGES"}:
-            continue
-        rhead = _normalize_sha(
-            review.get("reviewed_head_sha")
-            or review.get("commit_id")
-            or review.get("head_sha")
-        )
-        if rhead != head:
-            continue
-        matches.append({**review, "verdict": verdict, "reviewed_head_sha": rhead})
-    if not matches:
-        return None
-    return matches[-1]
-
-
-def find_newest_nonterminal_lease(
-    comments: list[dict],
-    *,
-    pr_number: int,
-    now: datetime | None = None,
-    include_expired: bool = True,
-) -> dict[str, Any] | None:
-    """Newest non-terminal lease marker, optionally including expired ones (#691).
-
-    Unlike ``find_active_reviewer_lease``, this retains expired non-terminal
-    markers so diagnosis/cleanup can still name the obsolete lease after
-    ``expires_at`` (comment-backed ledger; no control-plane ``lease_id``).
-    """
-    now = now or datetime.now(timezone.utc)
-    entries = list(reversed(_lease_entries(comments, pr_number=pr_number)))
-    for entry in entries:
-        phase = (entry.get("phase") or "").strip().lower()
-        if phase in _TERMINAL_PHASES:
-            # Newest terminal ends the ledger chain for active acquisition, but
-            # an older non-terminal is not "active". Stop at newest marker.
-            return None
-        expired = _lease_expired(entry, now=now)
-        if expired and not include_expired:
-            return None
-        lease = dict(entry)
-        lease["freshness"] = classify_lease_freshness(lease, now=now)
-        lease["expired"] = expired
-        return lease
-    return None
-
-
-def cleanup_confirmation_for_pr(pr_number: int) -> str:
-    return f"{CLEANUP_OBSOLETE_LEASE_CONFIRMATION_PREFIX}{int(pr_number)}"
-
-
-def assess_obsolete_reviewer_comment_lease_cleanup(
-    comments: list[dict],
-    *,
-    pr_number: int,
-    current_head_sha: str | None,
-    formal_reviews: list[dict] | None = None,
-    repo: str | None = None,
-    expected_repo: str | None = None,
-    requesting_session_id: str | None = None,
-    controller_recovery_authorized: bool = False,
-    worktree_exists: bool | None = None,
-    worktree_clean: bool | None = None,
-    worktree_has_unpreserved_work: bool | None = None,
-    owner_process_alive: bool | None = None,
-    owner_pid_observed: int | None = None,
-    requesting_pid: int | None = None,
-    current_head_review_in_progress: bool = False,
-    expected_lease_comment_id: int | None = None,
-    expected_session_id: str | None = None,
-    expected_leased_head: str | None = None,
-    confirmation: str | None = None,
-    apply: bool = False,
-    now: datetime | None = None,
-) -> dict[str, Any]:
-    """Guarded cleanup eligibility for obsolete comment-backed reviewer leases (#691).
-
-    Cleanup is never transfer/adoption/repoint and never uses PID equality as
-    ownership. Eligible only when evidence proves the lease is past expiry and/or
-    pinned to a superseded head with a completed formal terminal review, and
-    every safety boundary holds.
-    """
-    now = now or datetime.now(timezone.utc)
-    reasons: list[str] = []
-    fail_closed_reasons: list[str] = []
-    current_head = _normalize_sha(current_head_sha)
-    req_session = (requesting_session_id or "").strip()
-
-    lease = find_newest_nonterminal_lease(
-        comments, pr_number=pr_number, now=now, include_expired=True
-    )
-    if not lease:
-        # Fall back to active finder (unexpired only) for report consistency.
-        lease = find_active_reviewer_lease(comments, pr_number=pr_number, now=now)
-
-    classification = "no_lease"
-    cleanup_allowed = False
-    release_body: str | None = None
-    audit_comment_body: str | None = None
-    terminal_review = None
-    leased_head = None
-    head_superseded = False
-    past_expiry = False
-    expires_parseable = True
-
-    if not lease:
-        fail_closed_reasons.append(
-            f"PR #{pr_number}: no non-terminal comment-backed reviewer lease to clean"
-        )
-        classification = "no_lease"
-    else:
-        leased_head = _normalize_sha(lease.get("candidate_head"))
-        expires_raw = lease.get("expires_at")
-        expires_at = _parse_timestamp(expires_raw)
-        if expires_raw and expires_at is None:
-            expires_parseable = False
-            fail_closed_reasons.append(
-                "lease expires_at cannot be parsed or trusted (fail closed)"
-            )
-        past_expiry = bool(expires_at and expires_at <= now)
-        freshness = lease.get("freshness") or classify_lease_freshness(lease, now=now)
-        owner_session = (lease.get("session_id") or "").strip()
-        is_requesting_owner = bool(
-            req_session and owner_session and req_session == owner_session
-        )
-
-        # Identity pins (repo / PR / session / head / comment) must match when
-        # the caller supplies expected evidence.
-        lease_repo = (lease.get("repo") or "").strip()
-        exp_repo = (expected_repo or repo or "").strip()
-        if exp_repo and lease_repo and exp_repo != lease_repo:
-            fail_closed_reasons.append(
-                f"repository identity mismatch: lease repo={lease_repo!r} "
-                f"expected={exp_repo!r}"
-            )
-        if lease.get("pr_number") not in (None, pr_number):
-            fail_closed_reasons.append(
-                f"PR identity mismatch: lease pr={lease.get('pr_number')} "
-                f"requested={pr_number}"
-            )
-        if expected_lease_comment_id is not None:
-            cid = lease.get("comment_id")
-            if cid is None or int(cid) != int(expected_lease_comment_id):
-                fail_closed_reasons.append(
-                    "lease comment_id does not match expected evidence "
-                    f"(lease={cid}, expected={expected_lease_comment_id})"
-                )
-        if expected_session_id:
-            if owner_session != expected_session_id.strip():
-                fail_closed_reasons.append(
-                    "lease session_id does not match expected evidence "
-                    f"(lease={owner_session}, expected={expected_session_id})"
-                )
-        if expected_leased_head:
-            exp_head = _normalize_sha(expected_leased_head)
-            if not exp_head or exp_head != leased_head:
-                fail_closed_reasons.append(
-                    "leased head does not match expected evidence "
-                    f"(lease={leased_head}, expected={exp_head})"
-                )
-
-        # PID equality is never ownership proof (#691).
-        if (
-            owner_pid_observed is not None
-            and requesting_pid is not None
-            and int(owner_pid_observed) == int(requesting_pid)
-        ):
-            reasons.append(
-                "observed PID equals requesting PID but PID equality is NOT "
-                "ownership proof; ignored for authorization"
-            )
-
-        if is_requesting_owner:
-            fail_closed_reasons.append(
-                "requesting session owns the lease; use "
-                "gitea_release_reviewer_pr_lease (owner path), not non-owner cleanup"
-            )
-
-        if current_head_review_in_progress:
-            fail_closed_reasons.append(
-                "a current-head review submission is in progress; cleanup denied"
-            )
-
-        if not current_head:
-            fail_closed_reasons.append(
-                "current PR head SHA missing or unparseable (fail closed)"
-            )
-        if not leased_head:
-            fail_closed_reasons.append(
-                "lease candidate_head missing or unparseable (fail closed)"
-            )
-
-        head_superseded = bool(
-            current_head and leased_head and current_head != leased_head
-        )
-        head_matches_current = bool(
-            current_head and leased_head and current_head == leased_head
-        )
-
-        terminal_review = formal_terminal_review_for_head(
-            formal_reviews, leased_head=leased_head
-        )
-        has_terminal = terminal_review is not None
-
-        # Worktree safety.
-        if worktree_has_unpreserved_work is True or worktree_clean is False:
-            fail_closed_reasons.append(
-                "lease worktree is dirty or has unpreserved work; cleanup denied"
-            )
-        if (
-            worktree_exists is True
-            and worktree_clean is None
-            and worktree_has_unpreserved_work is None
-        ):
-            # Unknown cleanliness when path exists — fail closed.
-            fail_closed_reasons.append(
-                "lease worktree exists but cleanliness is unknown (fail closed)"
-            )
-
-        # Classification matrix (#691).
-        if head_matches_current and freshness in {"active", "stale_warning"}:
-            classification = "foreign_active_current_head"
-            fail_closed_reasons.append(
-                "genuinely active foreign lease on the current PR head; "
-                "cleanup denied (fail closed)"
-            )
-        elif head_matches_current and past_expiry:
-            classification = "foreign_expired_current_head"
-            # Expired on current head: owner-missing / orphan path may apply.
-            if owner_process_alive is True:
-                fail_closed_reasons.append(
-                    "lease expired on current head but owner process still alive "
-                    "with plausible activity; cleanup denied"
-                )
-            elif worktree_clean is False:
-                fail_closed_reasons.append(
-                    "owner process absent but worktree dirty; cleanup denied"
-                )
-            elif owner_process_alive is False and (
-                worktree_clean is True or worktree_exists is False
-            ):
-                if controller_recovery_authorized:
-                    classification = "orphaned_owner_missing"
-                else:
-                    fail_closed_reasons.append(
-                        "controller/recovery capability not authorized for orphan cleanup"
-                    )
-            else:
-                fail_closed_reasons.append(
-                    "insufficient orphan evidence for expired current-head lease "
-                    "(need owner_process_alive=false and clean/absent worktree)"
-                )
-        elif head_superseded and has_terminal and past_expiry:
-            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:
-            classification = "ambiguous_conflicting_evidence"
-            fail_closed_reasons.append(
-                "lease head is superseded but no formal terminal review exists "
-                "for the leased head (fail closed)"
-            )
-        elif not head_superseded and not past_expiry and freshness in {
-            "active", "stale_warning"
-        }:
-            classification = "foreign_active_current_head"
-            fail_closed_reasons.append(
-                "foreign lease still active on current head; wait or use owner release"
-            )
-        elif past_expiry and not head_superseded:
-            classification = "foreign_expired_current_head"
-        else:
-            classification = "ambiguous_conflicting_evidence"
-            fail_closed_reasons.append(
-                "lease evidence is ambiguous; cleanup denied (fail closed)"
-            )
-
-        # Recent owner progress on a non-superseded active lease blocks cleanup.
-        minutes = _minutes_since_activity(lease, now=now)
-        if (
-            head_matches_current
-            and freshness == "active"
-            and minutes is not None
-            and minutes < STALE_WARNING_MINUTES
-        ):
-            fail_closed_reasons.append(
-                "owner session has recent authenticated progress on current head; "
-                "cleanup denied"
-            )
-
-        # Authority + confirmation for apply.
-        if not controller_recovery_authorized:
-            if classification in {
-                "foreign_completed_superseded_head",
-                "foreign_expired_superseded_head",
-                "foreign_expired_current_head",
-                "orphaned_owner_missing",
-            }:
-                fail_closed_reasons.append(
-                    "controller/recovery capability required "
-                    "(controller_recovery_authorized=true)"
-                )
-
-        expected_conf = cleanup_confirmation_for_pr(pr_number)
-        if apply:
-            if (confirmation or "").strip() != expected_conf:
-                fail_closed_reasons.append(
-                    f"confirmation must equal exactly {expected_conf!r}"
-                )
-
-        # Superseded + terminal path does not require past_expiry (AC1).
-        eligible_class = classification in {
-            "foreign_completed_superseded_head",
-            "foreign_expired_superseded_head",
-            "orphaned_owner_missing",
-            "foreign_expired_current_head",
-        }
-        # foreign_expired_current_head needs orphan/clean worktree + authority.
-        if classification == "foreign_expired_current_head":
-            if worktree_clean is not True and worktree_exists is not False:
-                if "worktree" not in " ".join(fail_closed_reasons):
-                    fail_closed_reasons.append(
-                        "expired current-head cleanup requires proven-clean "
-                        "worktree or absent worktree"
-                    )
-            if owner_process_alive is True:
-                fail_closed_reasons.append(
-                    "owner process still alive on expired current-head lease"
-                )
-
-        cleanup_allowed = (
-            eligible_class
-            and expires_parseable
-            and not fail_closed_reasons
-            and not is_requesting_owner
-        )
-
-        if cleanup_allowed:
-            release_body = format_lease_body(
-                repo=lease.get("repo") or exp_repo or "",
-                pr_number=pr_number,
-                issue_number=lease.get("issue_number"),
-                reviewer_identity=lease.get("reviewer_identity") or "",
-                profile=lease.get("profile") or "unknown",
-                session_id=owner_session or "unknown",
-                worktree=lease.get("worktree") or "",
-                phase="released",
-                candidate_head=leased_head,
-                target_branch=lease.get("target_branch") or "master",
-                target_branch_sha=lease.get("target_branch_sha"),
-                last_activity=now,
-                blocker="obsolete-superseded-or-expired-lease",
-            )
-            audit_comment_body = (
-                "## Canonical obsolete reviewer lease cleanup (#691)\n\n"
-                f"- pr: #{pr_number}\n"
-                f"- lease_comment_id: {lease.get('comment_id')}\n"
-                f"- session_id: {owner_session}\n"
-                f"- leased_head: {leased_head}\n"
-                f"- current_head: {current_head}\n"
-                f"- expires_at: {lease.get('expires_at')}\n"
-                f"- classification: {classification}\n"
-                f"- terminal_review_verdict: "
-                f"{(terminal_review or {}).get('verdict')}\n"
-                f"- tool: {CLEANUP_OBSOLETE_LEASE_TOOL}\n"
-                "- action: posted terminal phase=released lease marker; "
-                "did not transfer validation, decision, or workflow proof; "
-                "did not repoint lease to the new head; did not delete history\n"
-            )
-            reasons.append(
-                f"cleanup eligible ({classification}); post terminal released "
-                "marker via sanctioned tool"
-            )
-        else:
-            reasons.extend(fail_closed_reasons)
-
-    return {
-        "pr_number": pr_number,
-        "classification": classification,
-        "blocker_kind": classification if not cleanup_allowed else "none",
-        "cleanup_allowed": cleanup_allowed,
-        "mutation_allowed": False,  # never grants review mutation
-        "mutation_eligibility": "prohibited" if not cleanup_allowed else "cleanup_only",
-        "exact_next_action": (
-            NEXT_ACTION_CLEANUP_OBSOLETE_LEASE
-            if cleanup_allowed
-            else (
-                NEXT_ACTION_WAIT
-                if classification
-                in {
-                    "foreign_active_current_head",
-                    "ambiguous_conflicting_evidence",
-                }
-                else NEXT_ACTION_OPERATOR_AUTHORIZED_CLEANUP
-            )
-        ),
-        "cleanup_tool": CLEANUP_OBSOLETE_LEASE_TOOL,
-        "required_confirmation": cleanup_confirmation_for_pr(pr_number),
-        "leased_head": leased_head,
-        "current_head": current_head,
-        "head_superseded": head_superseded,
-        "past_expiry": past_expiry,
-        "expires_at": (lease or {}).get("expires_at") if lease else None,
-        "expires_parseable": expires_parseable,
-        "terminal_review": terminal_review,
-        "terminal_review_present": terminal_review is not None,
-        "worktree_state": {
-            "path": (lease or {}).get("worktree") if lease else None,
-            "exists": worktree_exists,
-            "clean": worktree_clean,
-            "has_unpreserved_work": worktree_has_unpreserved_work,
-        },
-        "owner_session_evidence": {
-            "session_id": (lease or {}).get("session_id") if lease else None,
-            "reviewer_identity": (
-                (lease or {}).get("reviewer_identity") if lease else None
-            ),
-            "process_alive": owner_process_alive,
-            "pid_observed": owner_pid_observed,
-            "pid_is_not_ownership_proof": True,
-            "requesting_session_is_owner": bool(
-                lease
-                and req_session
-                and (lease.get("session_id") or "").strip() == req_session
-            ),
-        },
-        "active_lease": lease,
-        "release_body": release_body,
-        "audit_comment_body": audit_comment_body,
-        "controller_recovery_authorized": controller_recovery_authorized,
-        "reasons": reasons if reasons else fail_closed_reasons,
-        "fail_closed_reasons": fail_closed_reasons,
-        "forbidden": [
-            "manual comment deletion",
-            "database edits",
-            "mtime manipulation",
-            "PID-based ownership",
-            "direct session-state seeding",
-            "lease stealing or adoption",
-            "repointing old lease to new head",
-            "transfer of validation or decision state",
-            "worktree reuse by replacement reviewer",
-        ],
-    }
-
-
 def diagnose_reviewer_pr_lease_handoff(
     comments: list[dict],
     *,
@@ -1050,50 +561,40 @@ def diagnose_reviewer_pr_lease_handoff(
     env_bound_worktree: str | None = None,
     instructed_session_id: str | None = None,
     instructed_comment_id: int | None = None,
-    current_head_sha: str | None = None,
-    formal_reviews: list[dict] | None = None,
-    worktree_exists: bool | None = None,
-    worktree_clean: bool | None = None,
-    owner_process_alive: bool | None = None,
     now: datetime | None = None,
 ) -> dict[str, Any]:
-    """Classify open-PR reviewer lease handoff and emit a canonical next action (#599, #691).
+    """Classify open-PR reviewer lease handoff and emit a canonical next action (#599).
 
     Read-only diagnosis. Never steals, releases, or adopts a foreign lease.
     Fail-closed acquisition rules for active foreign leases remain intact.
 
     Returns a structured diagnosis with:
-    - classification (including #691 superseded/expired distinctions)
-    - next_action (including cleanup_obsolete_reviewer_comment_lease)
+    - classification
+    - next_action (one of wait / resume_exact_owner_session /
+      release_expired_lease / operator_authorized_cleanup /
+      repair_worktree_binding / acquire)
     - active_lease identity fields when present
     - worktree_binding match result
     - instructed-lease mismatch flags
-    - cleanup tool/confirmation when cleanup-eligible
     """
     now = now or datetime.now(timezone.utc)
     session_id = (current_session_id or "").strip()
     identity = (current_reviewer_identity or "").strip()
     instructed_sid = (instructed_session_id or "").strip()
     reasons: list[str] = []
-    current_head = _normalize_sha(current_head_sha)
 
     active = find_active_reviewer_lease(comments, pr_number=pr_number, now=now)
-    # Also surface expired non-terminal newest marker for #691 diagnosis.
-    newest_nt = find_newest_nonterminal_lease(
-        comments, pr_number=pr_number, now=now, include_expired=True
-    )
-    lease_for_class = active or newest_nt
     session_lease = get_session_lease()
 
     # Worktree binding: env-bound vs proposed vs active lease worktree.
     env_wt = _norm_path(env_bound_worktree)
     prop_wt = _norm_path(proposed_worktree)
-    lease_wt = _norm_path((lease_for_class or {}).get("worktree") if lease_for_class else None)
+    lease_wt = _norm_path((active or {}).get("worktree") if active else None)
     binding_mismatch = False
     binding_details: dict[str, Any] = {
         "env_bound_worktree": env_bound_worktree or None,
         "proposed_worktree": proposed_worktree or None,
-        "lease_worktree": (lease_for_class or {}).get("worktree") if lease_for_class else None,
+        "lease_worktree": (active or {}).get("worktree") if active else None,
         "match": True,
     }
     paths = [p for p in (env_wt, prop_wt, lease_wt) if p]
@@ -1107,13 +608,13 @@ def diagnose_reviewer_pr_lease_handoff(
     # Instructed lease gone while a different lease is active (PR #592-style).
     instructed_missing_with_replacement = False
     if instructed_sid or instructed_comment_id is not None:
-        if not lease_for_class:
+        if not active:
             reasons.append(
                 "instructed lease is gone and no active replacement lease remains"
             )
         else:
-            owner = (lease_for_class.get("session_id") or "").strip()
-            cid = lease_for_class.get("comment_id")
+            owner = (active.get("session_id") or "").strip()
+            cid = active.get("comment_id")
             sid_mismatch = bool(instructed_sid and owner and owner != instructed_sid)
             cid_mismatch = (
                 instructed_comment_id is not None
@@ -1130,94 +631,35 @@ def diagnose_reviewer_pr_lease_handoff(
     # Classification + next_action.
     classification = "no_lease"
     next_action = NEXT_ACTION_ACQUIRE
-    cleanup_hint: dict[str, Any] | None = None
 
-    if lease_for_class:
-        owner = (lease_for_class.get("session_id") or "").strip()
-        freshness = lease_for_class.get("freshness") or classify_lease_freshness(
-            lease_for_class, now=now
+    if active:
+        owner = (active.get("session_id") or "").strip()
+        freshness = active.get("freshness") or classify_lease_freshness(
+            active, now=now
         )
-        owner_identity = (lease_for_class.get("reviewer_identity") or "").strip()
+        owner_identity = (active.get("reviewer_identity") or "").strip()
         is_own = bool(session_id and owner and owner == session_id)
         # Same identity alone is NOT ownership for resume; session_id must match.
         same_identity = bool(
             identity and owner_identity and identity == owner_identity
         )
-        leased_head = _normalize_sha(lease_for_class.get("candidate_head"))
-        head_superseded = bool(
-            current_head and leased_head and current_head != leased_head
-        )
-        head_current = bool(
-            current_head and leased_head and current_head == leased_head
-        )
-        past_expiry = bool(lease_for_class.get("expired")) or freshness == "expired"
-        if not past_expiry:
-            past_expiry = _lease_expired(lease_for_class, now=now)
-        terminal = formal_terminal_review_for_head(
-            formal_reviews, leased_head=leased_head
-        )
 
-        if is_own and freshness in {"active", "stale_warning"} and not past_expiry:
+        if is_own and freshness in {"active", "stale_warning"}:
             classification = "own_active"
             next_action = NEXT_ACTION_RESUME_EXACT_OWNER_SESSION
-        elif is_own and (freshness in {"reclaimable", "expired"} or past_expiry):
+        elif is_own and freshness in {"reclaimable", "expired"}:
             classification = "own_expired"
             next_action = NEXT_ACTION_RELEASE_EXPIRED_LEASE
             reasons.append(
                 f"own lease freshness is '{freshness}'; release via "
                 "gitea_release_reviewer_pr_lease then re-acquire"
             )
-        elif not is_own and head_superseded and terminal and past_expiry:
-            classification = "foreign_expired_superseded_head"
-            next_action = NEXT_ACTION_CLEANUP_OBSOLETE_LEASE
-            reasons.append(
-                "foreign lease expired and pinned to superseded head with "
-                "formal terminal review; use guarded obsolete-lease cleanup"
-            )
-        elif not is_own and head_superseded and terminal and not past_expiry:
-            classification = "foreign_completed_superseded_head"
-            next_action = NEXT_ACTION_CLEANUP_OBSOLETE_LEASE
-            reasons.append(
-                "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:
-            classification = "ambiguous_conflicting_evidence"
-            next_action = NEXT_ACTION_WAIT
-            reasons.append(
-                "lease head superseded but no formal terminal review for leased "
-                "head; fail closed / wait (no indefinite steal)"
-            )
-        elif not is_own and head_current and past_expiry:
-            classification = "foreign_expired_current_head"
-            if owner_process_alive is False and worktree_clean is True:
-                classification = "orphaned_owner_missing"
-                next_action = NEXT_ACTION_CLEANUP_OBSOLETE_LEASE
-                reasons.append(
-                    "foreign expired lease on current head; owner process absent "
-                    "and worktree clean — guarded orphan cleanup eligible"
-                )
-            elif owner_process_alive is False and worktree_clean is False:
-                classification = "ambiguous_conflicting_evidence"
-                next_action = NEXT_ACTION_WAIT
-                reasons.append(
-                    "owner process absent but worktree dirty; cleanup denied"
-                )
-            else:
-                next_action = NEXT_ACTION_CLEANUP_OBSOLETE_LEASE
-                reasons.append(
-                    "foreign expired lease on current head; use guarded cleanup "
-                    "when worktree/process evidence permits"
-                )
-        elif not is_own and freshness in {"active", "stale_warning"} and not past_expiry:
-            classification = (
-                "foreign_active_current_head" if head_current or not current_head
-                else "foreign_active"
-            )
+        elif not is_own and freshness in {"active", "stale_warning"}:
+            classification = "foreign_active"
             next_action = NEXT_ACTION_WAIT
             reasons.append(
                 f"foreign active reviewer lease (session_id={owner}, "
-                f"phase={lease_for_class.get('phase')}, freshness={freshness}); "
+                f"phase={active.get('phase')}, freshness={freshness}); "
                 "do not submit; do not steal"
             )
             if same_identity:
@@ -1232,12 +674,11 @@ def diagnose_reviewer_pr_lease_handoff(
                 f"foreign reclaimable lease (session_id={owner}); clear only via "
                 "sanctioned gitea_release_reviewer_pr_lease when reclaimable"
             )
-        elif not is_own and (freshness == "expired" or past_expiry):
+        elif not is_own and freshness == "expired":
             classification = "foreign_expired"
             next_action = NEXT_ACTION_RELEASE_EXPIRED_LEASE
             reasons.append(
-                f"foreign expired lease (session_id={owner}); use sanctioned release "
-                f"or {CLEANUP_OBSOLETE_LEASE_TOOL}"
+                f"foreign expired lease (session_id={owner}); use sanctioned release"
             )
         else:
             classification = "foreign_active"
@@ -1250,26 +691,13 @@ def diagnose_reviewer_pr_lease_handoff(
         if instructed_missing_with_replacement and classification.startswith(
             "foreign"
         ):
-            # Preserve #691 cleanup path when superseded/expired; otherwise
-            # keep the PR #592 instructed-missing label.
-            if next_action != NEXT_ACTION_CLEANUP_OBSOLETE_LEASE:
-                classification = "instructed_lease_missing_with_replacement"
-                if next_action == NEXT_ACTION_WAIT:
-                    reasons.append(
-                        "replacement foreign lease is active — wait; "
-                        "operator_authorized_cleanup only with explicit operator authority"
-                    )
-
-        if next_action == NEXT_ACTION_CLEANUP_OBSOLETE_LEASE:
-            cleanup_hint = {
-                "cleanup_tool": CLEANUP_OBSOLETE_LEASE_TOOL,
-                "required_confirmation": cleanup_confirmation_for_pr(pr_number),
-                "controller_recovery_authorized_required": True,
-                "leased_head": leased_head,
-                "current_head": current_head,
-                "expires_at": lease_for_class.get("expires_at"),
-                "terminal_review_verdict": (terminal or {}).get("verdict"),
-            }
+            classification = "instructed_lease_missing_with_replacement"
+            # Foreign active still means wait; reclaimable still release.
+            if next_action == NEXT_ACTION_WAIT:
+                reasons.append(
+                    "replacement foreign lease is active — wait; "
+                    "operator_authorized_cleanup only with explicit operator authority"
+                )
     else:
         classification = "no_lease"
         next_action = NEXT_ACTION_ACQUIRE
@@ -1292,55 +720,29 @@ def diagnose_reviewer_pr_lease_handoff(
             )
 
     lease_summary = None
-    if lease_for_class:
+    if active:
         lease_summary = {
-            "comment_id": lease_for_class.get("comment_id"),
-            "session_id": lease_for_class.get("session_id"),
-            "phase": lease_for_class.get("phase"),
-            "candidate_head": lease_for_class.get("candidate_head"),
-            "expires_at": lease_for_class.get("expires_at"),
-            "last_activity": lease_for_class.get("last_activity"),
-            "freshness": lease_for_class.get("freshness")
-            or classify_lease_freshness(lease_for_class, now=now),
-            "reviewer_identity": lease_for_class.get("reviewer_identity"),
-            "profile": lease_for_class.get("profile"),
-            "worktree": lease_for_class.get("worktree"),
-            "blocker": lease_for_class.get("blocker"),
+            "comment_id": active.get("comment_id"),
+            "session_id": active.get("session_id"),
+            "phase": active.get("phase"),
+            "candidate_head": active.get("candidate_head"),
+            "expires_at": active.get("expires_at"),
+            "last_activity": active.get("last_activity"),
+            "freshness": active.get("freshness")
+            or classify_lease_freshness(active, now=now),
+            "reviewer_identity": active.get("reviewer_identity"),
+            "profile": active.get("profile"),
+            "worktree": active.get("worktree"),
+            "blocker": active.get("blocker"),
         }
 
     return {
         "pr_number": pr_number,
         "classification": classification,
-        "blocker_kind": classification,
         "next_action": next_action,
-        "exact_next_action": next_action,
         "active_lease": lease_summary,
         "session_lease": session_lease,
         "worktree_binding": binding_details,
-        "leased_head": (lease_summary or {}).get("candidate_head"),
-        "current_head": current_head,
-        "expires_at": (lease_summary or {}).get("expires_at"),
-        "terminal_review_state": (
-            formal_terminal_review_for_head(
-                formal_reviews,
-                leased_head=(lease_summary or {}).get("candidate_head"),
-            )
-            if lease_summary
-            else None
-        ),
-        "worktree_state": {
-            "exists": worktree_exists,
-            "clean": worktree_clean,
-            "path": (lease_summary or {}).get("worktree"),
-        },
-        "owner_session_evidence": {
-            "session_id": (lease_summary or {}).get("session_id"),
-            "process_alive": owner_process_alive,
-            "pid_is_not_ownership_proof": True,
-        },
-        "cleanup": cleanup_hint,
-        "cleanup_tool": (cleanup_hint or {}).get("cleanup_tool"),
-        "required_confirmation": (cleanup_hint or {}).get("required_confirmation"),
         "instructed_session_id": instructed_session_id,
         "instructed_comment_id": instructed_comment_id,
         "instructed_lease_missing_with_replacement": instructed_missing_with_replacement,
@@ -1349,19 +751,6 @@ def diagnose_reviewer_pr_lease_handoff(
             and not binding_mismatch
             and bool(session_lease)
         ),
-        "mutation_eligibility": (
-            "allowed"
-            if (
-                next_action == NEXT_ACTION_RESUME_EXACT_OWNER_SESSION
-                and not binding_mismatch
-                and bool(session_lease)
-            )
-            else (
-                "cleanup_only"
-                if next_action == NEXT_ACTION_CLEANUP_OBSOLETE_LEASE
-                else "prohibited"
-            )
-        ),
         "reasons": reasons,
         "forbidden": [
             "manual lock deletion",
@@ -1369,8 +758,5 @@ def diagnose_reviewer_pr_lease_handoff(
             "mtime manipulation",
             "direct _SESSION_LEASE seeding",
             "silent foreign lease steal",
-            "PID-based ownership",
-            "repointing old lease to new head",
-            "transfer of validation or decision state",
         ],
     }

stale_review_decision_lock.py

@@ -1,4 +1,4 @@
-"""Stale / moot #332 review-decision lock detection and cleanup policy (#594/#620/#693).
+"""Stale / moot #332 review-decision lock detection and cleanup policy (#594/#620).
 
 #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,11 +12,6 @@ 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.
@@ -85,45 +80,17 @@ 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/#693).
+    """True when prior live mutations block a new decision for PR+head (#620).
 
     Historical terminals on a **different head of the same PR** do not block.
-    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.
+    Any prior mutation on another PR, the same head, or without a comparable
+    head SHA fails closed (blocks).
 
     Head resolution uses ``mutation_head_sha(m, lock)`` so pre-#620 ledgers
     that only stored ``ready_expected_head_sha`` still compare correctly
@@ -131,9 +98,7 @@ def prior_live_mutations_block_boundary(
     """
     if not lock:
         return False
-    if correction_applies(
-        lock, pr_number=pr_number, expected_head_sha=expected_head_sha
-    ):
+    if lock.get("correction_authorized"):
         return False
     prior = list(lock.get("live_mutations") or [])
     if not prior:
@@ -143,14 +108,10 @@ 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 (
-            target
+            m_pr == pr_number
+            and target
             and m_head
             and not heads_equal(m_head, target)
         ):
@@ -167,34 +128,23 @@ 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/#693).
+    """Whether #332 hard-stop should yield for a new PR+head boundary (#620).
 
-    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).
+    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).
     """
     if operation not in ("mark_ready", "review", "resume"):
         return False
     if not lock:
         return False
-    if correction_applies(
-        lock, pr_number=pr_number, expected_head_sha=expected_head_sha
-    ):
+    if lock.get("correction_authorized"):
         return True
     last = last_terminal_mutation(lock)
     if last is None:
         return True
-    last_pr = last.get("pr_number")
-    if last_pr is None:
+    if last.get("pr_number") != pr_number:
         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:
@@ -488,397 +438,3 @@ def format_cleanup_audit_comment(audit: dict[str, Any]) -> str:
         "This path only clears a lock when the referenced PR is merged/closed.",
     ]
     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,
-    }

task_capability_map.py

@@ -76,16 +76,6 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
         "permission": "gitea.pr.comment",
         "role": "reviewer",
     },
-    # #691: guarded non-owner cleanup of obsolete comment-backed reviewer leases.
-    # Apply path posts lease release + audit comments (gitea.pr.comment).
-    "cleanup_obsolete_reviewer_comment_lease": {
-        "permission": "gitea.pr.comment",
-        "role": "reviewer",
-    },
-    "gitea_cleanup_obsolete_reviewer_comment_lease": {
-        "permission": "gitea.pr.comment",
-        "role": "reviewer",
-    },
     "blind_pr_queue_review": {
         "permission": "gitea.pr.review",
         "role": "reviewer",
@@ -117,23 +107,6 @@ 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",
-    },
     "delete_branch": {
         "permission": "gitea.branch.delete",
         "role": "author",

tests/test_audit_reconciliation_mode.py

@@ -27,7 +27,13 @@ from task_capability_map import required_permission, required_role
 
 DELETE_PROFILE = {
     "profile_name": "prgs-author-delete",
-    "allowed_operations": ["gitea.read", "gitea.branch.delete"],
+    "role": "author",
+    "allowed_operations": [
+        "gitea.read",
+        "gitea.pr.create",
+        "gitea.branch.push",
+        "gitea.branch.delete",
+    ],
     "forbidden_operations": [],
     "audit_label": "prgs-author-delete",
 }

tests/test_head_scoped_review_decision_lock.py

@@ -24,10 +24,6 @@ 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",
@@ -44,11 +40,9 @@ 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": mutations,
+        "live_mutations": list(mutations or []),
         "correction_authorized": correction,
-        "correction_reason": "test" if correction else None,
-        "correction_pr_number": corr_pr,
-        "correction_head_sha": None,
+        "correction_reason": None,
     }
 
 
@@ -180,13 +174,12 @@ class TestHardStopHeadScope(unittest.TestCase):
         self.assertTrue(reasons)
         self.assertIn("#332", reasons[0])
 
-    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."""
+    def test_rc_head_a_blocks_other_pr(self):
         _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.assertEqual(reasons, [])
+        self.assertTrue(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

@@ -1,599 +0,0 @@
-"""Guarded cleanup for obsolete comment-backed reviewer leases (#691).
-
-Reproduces PR #688 lease comment 10749 class of defect: completed
-REQUEST_CHANGES on head A, author pushes head B, foreign lease remains
-pinned to A (and after expiry still has no non-owner cleanup path that
-diagnosis can prescribe beyond indefinite wait).
-"""
-
-from __future__ import annotations
-
-import sys
-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 merger_lease_adoption as mla  # noqa: E402
-import reviewer_pr_lease as leases  # noqa: E402
-
-# PR #688 / comment 10749 reproduction fixtures
-HEAD_A = "c7a444eb4b41cf916fdbd20a4999ffd78af496d0"
-HEAD_B = "4a6357800364718a27a36ebc73578d4b929ff4aa"
-SESSION_FOREIGN = "25883-7d4c6e6ebd53"
-COMMENT_10749 = 10749
-REPO = "Scaled-Tech-Consulting/Gitea-Tools"
-PR = 688
-ISSUE = 687
-EXPIRES_10749 = "2026-07-13T04:23:00Z"
-
-
-def _utc(dt: datetime) -> datetime:
-    return dt if dt.tzinfo else dt.replace(tzinfo=timezone.utc)
-
-
-def _lease_comment(
-    *,
-    pr_number: int = PR,
-    session_id: str = SESSION_FOREIGN,
-    phase: str = "claimed",
-    candidate_head: str = HEAD_A,
-    comment_id: int = COMMENT_10749,
-    last_activity: datetime | None = None,
-    expires_at: datetime | None = None,
-    worktree: str = "branches/review-pr688-feat-issue-687-reconciler-branch-delete",
-    repo: str = REPO,
-) -> dict:
-    now = last_activity or datetime.now(timezone.utc)
-    body = leases.format_lease_body(
-        repo=repo,
-        pr_number=pr_number,
-        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="b" * 40,
-        last_activity=now,
-        expires_at=expires_at,
-    )
-    return {
-        "id": comment_id,
-        "body": body,
-        "user": {"login": "sysadmin"},
-        "author": "sysadmin",
-        "created_at": now.isoformat(),
-        "updated_at": now.isoformat(),
-    }
-
-
-def _reviews_for_head(head: str, verdict: str = "REQUEST_CHANGES") -> list[dict]:
-    return [
-        {
-            "verdict": verdict,
-            "reviewed_head_sha": head,
-            "dismissed": False,
-            "reviewer": "sysadmin",
-        }
-    ]
-
-
-def _assess(
-    comments,
-    *,
-    current_head=HEAD_B,
-    formal_reviews=None,
-    controller=True,
-    worktree_exists=True,
-    worktree_clean=True,
-    owner_process_alive=False,
-    requesting_session="fresh-controller",
-    apply=False,
-    confirmation=None,
-    **kwargs,
-):
-    return leases.assess_obsolete_reviewer_comment_lease_cleanup(
-        comments,
-        pr_number=PR,
-        current_head_sha=current_head,
-        formal_reviews=formal_reviews if formal_reviews is not None else _reviews_for_head(HEAD_A),
-        repo=REPO,
-        expected_repo=REPO,
-        requesting_session_id=requesting_session,
-        controller_recovery_authorized=controller,
-        worktree_exists=worktree_exists,
-        worktree_clean=worktree_clean,
-        worktree_has_unpreserved_work=not worktree_clean if worktree_exists else False,
-        owner_process_alive=owner_process_alive,
-        owner_pid_observed=kwargs.get("owner_pid_observed"),
-        requesting_pid=kwargs.get("requesting_pid", 99999),
-        current_head_review_in_progress=kwargs.get(
-            "current_head_review_in_progress", False
-        ),
-        expected_lease_comment_id=kwargs.get("expected_lease_comment_id"),
-        expected_session_id=kwargs.get("expected_session_id"),
-        expected_leased_head=kwargs.get("expected_leased_head"),
-        confirmation=confirmation
-        if confirmation is not None
-        else (leases.cleanup_confirmation_for_pr(PR) if apply else ""),
-        apply=apply,
-        now=kwargs.get("now"),
-    )
-
-
-class TestIssue691SupersededHeadCleanup(unittest.TestCase):
-    def setUp(self):
-        leases.clear_session_lease()
-
-    def test_1_request_changes_then_push_expired_cleanup_succeeds(self):
-        """Completed REQUEST_CHANGES on A, push B, lease A expires → cleanup OK."""
-        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(
-            last_activity=expires - timedelta(hours=2),
-            expires_at=expires,
-        )
-        result = _assess(
-            [claim],
-            formal_reviews=_reviews_for_head(HEAD_A, "REQUEST_CHANGES"),
-            now=now,
-            apply=True,
-        )
-        self.assertTrue(result["cleanup_allowed"], result["reasons"])
-        self.assertEqual(result["classification"], "foreign_expired_superseded_head")
-        self.assertEqual(
-            result["exact_next_action"], leases.NEXT_ACTION_CLEANUP_OBSOLETE_LEASE
-        )
-        self.assertIn("phase: released", result["release_body"])
-        self.assertIn("obsolete-superseded-or-expired-lease", result["release_body"])
-        self.assertIsNotNone(result["audit_comment_body"])
-        self.assertEqual(result["cleanup_tool"], leases.CLEANUP_OBSOLETE_LEASE_TOOL)
-
-    def test_2_approve_then_push_cleanup_policy(self):
-        """Completed APPROVE on A, push B → cleanup eligible (completed superseded)."""
-        now = datetime(2026, 7, 13, 1, 0, 0, tzinfo=timezone.utc)
-        expires = datetime(2026, 7, 13, 4, 23, 0, tzinfo=timezone.utc)  # not yet expired
-        claim = _lease_comment(
-            last_activity=now - timedelta(minutes=10),
-            expires_at=expires,
-        )
-        result = _assess(
-            [claim],
-            formal_reviews=_reviews_for_head(HEAD_A, "APPROVED"),
-            now=now,
-        )
-        self.assertTrue(result["cleanup_allowed"], result["reasons"])
-        self.assertEqual(result["classification"], "foreign_completed_superseded_head")
-
-    def test_3_active_current_head_cleanup_denied(self):
-        now = datetime.now(timezone.utc)
-        claim = _lease_comment(
-            candidate_head=HEAD_B,
-            last_activity=now - timedelta(minutes=5),
-            expires_at=now + timedelta(hours=2),
-        )
-        result = _assess(
-            [claim],
-            current_head=HEAD_B,
-            formal_reviews=_reviews_for_head(HEAD_B, "REQUEST_CHANGES"),
-            now=now,
-        )
-        self.assertFalse(result["cleanup_allowed"])
-        self.assertEqual(result["classification"], "foreign_active_current_head")
-
-    def test_4_foreign_owner_recent_progress_denied(self):
-        now = datetime.now(timezone.utc)
-        claim = _lease_comment(
-            candidate_head=HEAD_B,
-            last_activity=now - timedelta(minutes=2),
-            expires_at=now + timedelta(hours=2),
-        )
-        result = _assess(
-            [claim],
-            current_head=HEAD_B,
-            formal_reviews=[],
-            owner_process_alive=True,
-            now=now,
-        )
-        self.assertFalse(result["cleanup_allowed"])
-        self.assertTrue(
-            any("recent authenticated progress" in r for r in result["reasons"])
-            or any("current PR head" in r for r in result["reasons"])
-        )
-
-    def test_5_owner_absent_worktree_dirty_denied(self):
-        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(
-            candidate_head=HEAD_B,
-            last_activity=expires - timedelta(hours=1),
-            expires_at=expires,
-        )
-        result = _assess(
-            [claim],
-            current_head=HEAD_B,
-            formal_reviews=[],
-            worktree_clean=False,
-            owner_process_alive=False,
-            now=now,
-        )
-        self.assertFalse(result["cleanup_allowed"])
-        self.assertTrue(any("dirty" in r for r in result["reasons"]))
-
-    def test_6_owner_absent_worktree_clean_orphan_ok(self):
-        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(
-            candidate_head=HEAD_B,
-            last_activity=expires - timedelta(hours=1),
-            expires_at=expires,
-        )
-        result = _assess(
-            [claim],
-            current_head=HEAD_B,
-            formal_reviews=[],
-            worktree_clean=True,
-            owner_process_alive=False,
-            now=now,
-        )
-        self.assertTrue(result["cleanup_allowed"], result["reasons"])
-        self.assertEqual(result["classification"], "orphaned_owner_missing")
-
-    def test_7_pid_reuse_not_ownership(self):
-        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],
-            now=now,
-            owner_pid_observed=4242,
-            requesting_pid=4242,
-        )
-        self.assertTrue(
-            any("PID equality" in r or "NOT ownership" in r for r in result["reasons"])
-            or result["owner_session_evidence"]["pid_is_not_ownership_proof"]
-        )
-        # Still eligible via superseded+terminal+expired despite matching PID.
-        self.assertTrue(result["cleanup_allowed"], result["reasons"])
-
-    def test_8_comment_backed_no_db_lease_id(self):
-        """Cleanup uses comment ledger only — no control-plane lease_id required."""
-        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))
-        # Explicitly no lease_id field anywhere.
-        self.assertNotIn("lease_id", claim)
-        result = _assess([claim], now=now)
-        self.assertTrue(result["cleanup_allowed"], result["reasons"])
-        self.assertEqual(result["active_lease"]["comment_id"], COMMENT_10749)
-        self.assertIsNone(result["active_lease"].get("lease_id"))
-
-    def test_9_cleanup_posts_durable_audit_bodies(self):
-        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], now=now, apply=True)
-        self.assertTrue(result["cleanup_allowed"])
-        self.assertIn("#691", result["audit_comment_body"])
-        self.assertIn(str(COMMENT_10749), result["audit_comment_body"])
-        self.assertIn(HEAD_A, result["audit_comment_body"])
-        self.assertIn(HEAD_B, result["audit_comment_body"])
-
-    def test_10_fresh_acquire_after_cleanup_marker(self):
-        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),
-            comment_id=COMMENT_10749,
-        )
-        assessed = _assess([claim], now=now, apply=True)
-        self.assertTrue(assessed["cleanup_allowed"])
-        release = {
-            "id": 99999,
-            "body": assessed["release_body"],
-            "user": {"login": "sysadmin"},
-        }
-        # Newest terminal marker ends active lease.
-        active = leases.find_active_reviewer_lease(
-            [claim, release], pr_number=PR, now=now
-        )
-        self.assertIsNone(active)
-        acq = leases.assess_acquire_lease(
-            [claim, release],
-            pr_number=PR,
-            reviewer_identity="sysadmin",
-            profile="prgs-reviewer",
-            session_id="fresh-reviewer-1",
-            repo=REPO,
-            issue_number=ISSUE,
-            worktree="branches/review-pr-688-fresh",
-            candidate_head=HEAD_B,
-            target_branch="master",
-            target_branch_sha="b" * 40,
-            now=now,
-        )
-        self.assertTrue(acq["acquire_allowed"], acq["reasons"])
-
-    def test_11_no_validation_state_transfer(self):
-        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], now=now, apply=True)
-        self.assertTrue(result["cleanup_allowed"])
-        # Release body keeps old candidate_head (no repoint).
-        self.assertIn(f"candidate_head: {HEAD_A}", result["release_body"])
-        self.assertNotIn(HEAD_B, result["release_body"].split("candidate_head:")[1].split("\n")[0])
-        self.assertIn("did not transfer validation", result["audit_comment_body"])
-        self.assertIn("did not repoint", result["audit_comment_body"])
-        # Session lease must remain unset by assessment (tool never seeds).
-        self.assertIsNone(leases.get_session_lease())
-
-    def test_12_repeated_cleanup_idempotent(self):
-        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))
-        first = _assess([claim], now=now, apply=True)
-        self.assertTrue(first["cleanup_allowed"])
-        release = {"id": 100001, "body": first["release_body"], "user": {"login": "x"}}
-        second = _assess([claim, release], now=now, apply=True)
-        self.assertFalse(second["cleanup_allowed"])
-        self.assertEqual(second["classification"], "no_lease")
-
-    def test_13_malformed_expiry_fails_closed(self):
-        claim = _lease_comment()
-        # Corrupt expires_at in the body.
-        claim["body"] = claim["body"].replace(
-            claim["body"].split("expires_at:")[1].split("\n")[0].strip(),
-            "not-a-timestamp",
-        )
-        result = _assess([claim], formal_reviews=_reviews_for_head(HEAD_A))
-        self.assertFalse(result["cleanup_allowed"])
-        self.assertFalse(result["expires_parseable"])
-        self.assertTrue(any("expires_at" in r for r in result["reasons"]))
-
-    def test_14_wrong_identity_evidence_fails_closed(self):
-        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))
-        bad_repo = _assess(
-            [claim],
-            now=now,
-            # force repo mismatch via expected_repo
-        )
-        # Patch via direct call with wrong expected_repo
-        bad = leases.assess_obsolete_reviewer_comment_lease_cleanup(
-            [claim],
-            pr_number=PR,
-            current_head_sha=HEAD_B,
-            formal_reviews=_reviews_for_head(HEAD_A),
-            expected_repo="Other-Org/Other-Repo",
-            requesting_session_id="fresh",
-            controller_recovery_authorized=True,
-            worktree_exists=True,
-            worktree_clean=True,
-            owner_process_alive=False,
-            now=now,
-        )
-        self.assertFalse(bad["cleanup_allowed"])
-        self.assertTrue(any("repository identity" in r for r in bad["reasons"]))
-
-        bad_pr = leases.assess_obsolete_reviewer_comment_lease_cleanup(
-            [claim],
-            pr_number=999,
-            current_head_sha=HEAD_B,
-            formal_reviews=_reviews_for_head(HEAD_A),
-            expected_repo=REPO,
-            requesting_session_id="fresh",
-            controller_recovery_authorized=True,
-            worktree_exists=True,
-            worktree_clean=True,
-            owner_process_alive=False,
-            expected_lease_comment_id=COMMENT_10749,
-            now=now,
-        )
-        self.assertFalse(bad_pr["cleanup_allowed"])
-
-        bad_head = _assess(
-            [claim],
-            now=now,
-            expected_leased_head="0" * 40,
-        )
-        self.assertFalse(bad_head["cleanup_allowed"])
-
-        bad_session = _assess(
-            [claim],
-            now=now,
-            expected_session_id="wrong-session",
-        )
-        self.assertFalse(bad_session["cleanup_allowed"])
-
-    def test_15_current_head_active_cannot_be_displaced(self):
-        now = datetime.now(timezone.utc)
-        claim = _lease_comment(
-            candidate_head=HEAD_B,
-            last_activity=now - timedelta(minutes=1),
-            expires_at=now + timedelta(hours=2),
-        )
-        result = _assess(
-            [claim],
-            current_head=HEAD_B,
-            formal_reviews=_reviews_for_head(HEAD_B),
-            now=now,
-            current_head_review_in_progress=True,
-        )
-        self.assertFalse(result["cleanup_allowed"])
-        # Acquire also blocked by foreign active lease.
-        acq = leases.assess_acquire_lease(
-            [claim],
-            pr_number=PR,
-            reviewer_identity="other",
-            profile="prgs-reviewer",
-            session_id="thief",
-            repo=REPO,
-            issue_number=ISSUE,
-            worktree="branches/steal",
-            candidate_head=HEAD_B,
-            target_branch="master",
-            target_branch_sha="b" * 40,
-            now=now,
-        )
-        self.assertFalse(acq["acquire_allowed"])
-
-    def test_regression_pr688_comment_10749_after_expiry(self):
-        """Exact 10749 reproduction after recorded expires_at 2026-07-13T04:23:00Z."""
-        now = datetime(2026, 7, 13, 4, 30, 0, tzinfo=timezone.utc)
-        expires = datetime.fromisoformat(EXPIRES_10749.replace("Z", "+00:00"))
-        claim = _lease_comment(
-            session_id=SESSION_FOREIGN,
-            candidate_head=HEAD_A,
-            comment_id=COMMENT_10749,
-            last_activity=expires - timedelta(hours=2),
-            expires_at=expires,
-        )
-        # Diagnosis must not prescribe indefinite wait-only for this case.
-        diag = leases.diagnose_reviewer_pr_lease_handoff(
-            [claim],
-            pr_number=PR,
-            current_session_id="fresh-reviewer",
-            current_reviewer_identity="sysadmin",
-            current_head_sha=HEAD_B,
-            formal_reviews=_reviews_for_head(HEAD_A, "REQUEST_CHANGES"),
-            worktree_exists=True,
-            worktree_clean=True,
-            owner_process_alive=False,
-            instructed_session_id=SESSION_FOREIGN,
-            instructed_comment_id=COMMENT_10749,
-            now=now,
-        )
-        self.assertEqual(diag["classification"], "foreign_expired_superseded_head")
-        self.assertEqual(
-            diag["next_action"], leases.NEXT_ACTION_CLEANUP_OBSOLETE_LEASE
-        )
-        self.assertEqual(diag["cleanup_tool"], leases.CLEANUP_OBSOLETE_LEASE_TOOL)
-        self.assertIn("CLEANUP OBSOLETE REVIEWER LEASE 688", diag["required_confirmation"])
-        self.assertEqual(diag["mutation_eligibility"], "cleanup_only")
-        self.assertFalse(diag["mutation_allowed"])
-
-        # Assessment still returns the lease as active/stale class of block for acquire
-        # when unexpired path... here it's expired so find_active is None; acquire free
-        # only after cleanup if somehow still active. With expired, find_active is None:
-        active = leases.find_active_reviewer_lease([claim], pr_number=PR, now=now)
-        self.assertIsNone(active)
-
-        # But superseded unexpired still blocks acquire and diagnosis points to cleanup:
-        unexpired_now = datetime(2026, 7, 13, 1, 0, 0, tzinfo=timezone.utc)
-        claim2 = _lease_comment(
-            session_id=SESSION_FOREIGN,
-            candidate_head=HEAD_A,
-            comment_id=COMMENT_10749,
-            last_activity=unexpired_now - timedelta(minutes=45),
-            expires_at=expires,
-        )
-        active2 = leases.find_active_reviewer_lease(
-            [claim2], pr_number=PR, now=unexpired_now
-        )
-        self.assertIsNotNone(active2)
-        self.assertEqual(active2["freshness"], "stale_warning")
-        diag2 = leases.diagnose_reviewer_pr_lease_handoff(
-            [claim2],
-            pr_number=PR,
-            current_session_id="fresh-reviewer",
-            current_reviewer_identity="sysadmin",
-            current_head_sha=HEAD_B,
-            formal_reviews=_reviews_for_head(HEAD_A, "REQUEST_CHANGES"),
-            worktree_exists=True,
-            worktree_clean=True,
-            now=unexpired_now,
-        )
-        self.assertEqual(diag2["classification"], "foreign_completed_superseded_head")
-        self.assertEqual(
-            diag2["next_action"], leases.NEXT_ACTION_CLEANUP_OBSOLETE_LEASE
-        )
-        acq = leases.assess_acquire_lease(
-            [claim2],
-            pr_number=PR,
-            reviewer_identity="sysadmin",
-            profile="prgs-reviewer",
-            session_id="fresh-reviewer",
-            repo=REPO,
-            issue_number=ISSUE,
-            worktree="branches/review-pr-688-new",
-            candidate_head=HEAD_B,
-            target_branch="master",
-            target_branch_sha="b" * 40,
-            now=unexpired_now,
-        )
-        self.assertFalse(acq["acquire_allowed"])
-
-    def test_missing_controller_auth_denied(self):
-        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], controller=False, now=now)
-        self.assertFalse(result["cleanup_allowed"])
-
-    def test_wrong_confirmation_denied_on_apply(self):
-        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], now=now, apply=True, confirmation="MERGE PR 688"
-        )
-        self.assertFalse(result["cleanup_allowed"])
-        self.assertTrue(any("confirmation" in r for r in result["reasons"]))
-
-    def test_owner_session_must_use_owner_release(self):
-        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],
-            now=now,
-            requesting_session=SESSION_FOREIGN,
-        )
-        self.assertFalse(result["cleanup_allowed"])
-        self.assertTrue(any("owns the lease" in r for r in result["reasons"]))
-
-    def test_superseded_without_terminal_review_denied(self):
-        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)
-        self.assertFalse(result["cleanup_allowed"])
-        self.assertEqual(result["classification"], "ambiguous_conflicting_evidence")
-
-
-class TestIssue691DiagnoseClassifications(unittest.TestCase):
-    def setUp(self):
-        leases.clear_session_lease()
-
-    def test_foreign_active_current_head_still_wait(self):
-        now = datetime.now(timezone.utc)
-        claim = _lease_comment(
-            candidate_head=HEAD_B,
-            last_activity=now - timedelta(minutes=5),
-            expires_at=now + timedelta(hours=1),
-        )
-        claim["id"] = 1
-        result = leases.diagnose_reviewer_pr_lease_handoff(
-            [claim],
-            pr_number=PR,
-            current_session_id="other",
-            current_reviewer_identity="sysadmin",
-            current_head_sha=HEAD_B,
-            formal_reviews=[],
-            now=now,
-        )
-        self.assertEqual(result["classification"], "foreign_active_current_head")
-        self.assertEqual(result["next_action"], leases.NEXT_ACTION_WAIT)
-        self.assertFalse(result["mutation_allowed"])
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/test_issue_693_review_decision_lock_recovery.py

@@ -1,427 +0,0 @@
-"""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_mcp_server.py

@@ -1425,10 +1425,16 @@ class TestReviewPR(unittest.TestCase):
 class TestDeleteBranch(unittest.TestCase):
 
     DELETE_PROFILE = {
-        "profile_name": "test-deleter",
-        "allowed_operations": ["gitea.read", "gitea.branch.delete"],
+        "profile_name": "test-author-deleter",
+        "role": "author",
+        "allowed_operations": [
+            "gitea.read",
+            "gitea.pr.create",
+            "gitea.branch.push",
+            "gitea.branch.delete",
+        ],
         "forbidden_operations": [],
-        "audit_label": "test-deleter",
+        "audit_label": "test-author-deleter",
     }
 
     @patch("mcp_server.get_profile", return_value=DELETE_PROFILE)
@@ -2540,18 +2546,15 @@ 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)
-        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,
-            )
+        r = gitea_authorize_review_correction(
+            prior_review_id=42,
+            prior_review_state="approve",
+            reason="typo",
+            operator_authorized=True,
+        )
         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
@@ -2565,7 +2568,6 @@ 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"]))
@@ -2576,22 +2578,10 @@ 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 {}
@@ -2604,7 +2594,6 @@ 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"]))
@@ -2662,7 +2651,6 @@ 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_migrate_profiles.py

@@ -92,14 +92,19 @@ class TestMigrateProfiles(unittest.TestCase):
         author = prgs_gitea["identities"]["author"]
         self.assertEqual(author["username"], "jcwalker3")
         self.assertEqual(author["auth"]["id"], "redacted-author-ref")
-        self.assertEqual(author["allowed_operations"], ["read", "comment"])
-        self.assertEqual(author["forbidden_operations"], ["approve", "merge"])
+        self.assertEqual(
+            author["allowed_operations"], ["gitea.read", "gitea.pr.comment"]
+        )
+        self.assertEqual(
+            author["forbidden_operations"],
+            ["gitea.pr.approve", "gitea.pr.merge"],
+        )
 
         reviewer = prgs_gitea["identities"]["reviewer"]
         self.assertEqual(reviewer["role"], "reviewer")
         self.assertEqual(reviewer["username"], "sysadmin")
         self.assertEqual(reviewer["auth"]["id"], "redacted-reviewer-ref")
-        self.assertIn("merge", reviewer["allowed_operations"])
+        self.assertIn("gitea.pr.merge", reviewer["allowed_operations"])
 
     def test_alias_generation(self):
         """Test that aliases are correctly generated to support old profile names."""
@@ -188,7 +193,7 @@ class TestMigrateProfiles(unittest.TestCase):
         self.assertNotIn("token", stdout_output.lower())
 
     def test_explicit_operations_are_preserved(self):
-        """Explicit v1 permissions must not be replaced by role defaults."""
+        """Explicit v1 permissions are canonicalized, not replaced by role defaults."""
         v1_data = json.loads(json.dumps(self.v1_content))
         v1_data["profiles"]["prgs-reviewer"]["allowed_operations"] = ["read"]
         v1_data["profiles"]["prgs-reviewer"]["forbidden_operations"] = ["merge"]
@@ -198,8 +203,8 @@ class TestMigrateProfiles(unittest.TestCase):
             v2_data["environments"]["prgs"]["services"]["gitea"]
             ["identities"]["reviewer"]
         )
-        self.assertEqual(reviewer["allowed_operations"], ["read"])
-        self.assertEqual(reviewer["forbidden_operations"], ["merge"])
+        self.assertEqual(reviewer["allowed_operations"], ["gitea.read"])
+        self.assertEqual(reviewer["forbidden_operations"], ["gitea.pr.merge"])
 
     def test_inferred_role_defaults_only_when_unambiguous(self):
         """Role defaults are allowed only for clear author/reviewer profiles."""
@@ -306,6 +311,171 @@ class TestMigrateProfiles(unittest.TestCase):
                 migrate_profiles.main()
             self.assertEqual(cm.exception.code, 1)
 
+    def test_reconciler_profile_migration(self):
+        """Legacy reconciler shorthands migrate to valid canonical operations."""
+        import gitea_config
+        import reconciler_profile
+
+        v1_data = {
+            "version": 1,
+            "profiles": {
+                "prgs-reconciler": {
+                    "base_url": "redacted-prgs-service",
+                    "username": "reconciler-agent",
+                    "auth": {"type": "keychain", "id": "reconciler-ref"},
+                    "execution_profile": "prgs-reconciler",
+                    "allowed_operations": [
+                        "read",
+                        "pr.close",
+                        "pr.comment",
+                        "issue.comment",
+                        "issue.close",
+                        "gitea.branch.delete",
+                    ],
+                    "forbidden_operations": [
+                        "merge",
+                        "approve",
+                        "review",
+                        "pr.create",
+                        "branch.push",
+                        "commit",
+                    ],
+                }
+            }
+        }
+        v2_data = migrate_profiles.migrate_v1_to_v2(v1_data)
+        reconciler = (
+            v2_data["environments"]["prgs"]["services"]["gitea"]
+            ["identities"]["reconciler"]
+        )
+        self.assertEqual(reconciler["role"], "reconciler")
+        allowed = reconciler["allowed_operations"]
+        forbidden = reconciler["forbidden_operations"]
+        # No invalid shorthand remains
+        for bad in ("pr.close", "pr.comment", "issue.close", "read", "merge"):
+            self.assertNotIn(bad, allowed)
+            self.assertNotIn(bad, forbidden)
+        for required in (
+            "gitea.read",
+            "gitea.pr.close",
+            "gitea.pr.comment",
+            "gitea.issue.comment",
+            "gitea.issue.close",
+            "gitea.branch.delete",
+        ):
+            self.assertIn(required, allowed)
+            # Production loader accepts every allowed op
+            self.assertEqual(
+                gitea_config.normalize_operation(required), required
+            )
+        self.assertEqual(v2_data["aliases"]["prgs-reconciler"], "prgs.gitea.reconciler")
+        assessment = reconciler_profile.assess_reconciler_profile(allowed, forbidden)
+        self.assertTrue(assessment["valid"])
+        self.assertTrue(migrate_profiles.validate_v2_data(v2_data))
+
+    def test_reconciler_profile_defaults(self):
+        """Reconciler defaults are fully canonical and loader-valid."""
+        import gitea_config
+        import reconciler_profile
+
+        v1_data = {
+            "version": 1,
+            "profiles": {
+                "prgs-reconciler": {
+                    "base_url": "redacted-prgs-service",
+                    "username": "reconciler-agent",
+                    "auth": {"type": "keychain", "id": "reconciler-ref"},
+                    "execution_profile": "prgs-reconciler",
+                }
+            }
+        }
+        v2_data = migrate_profiles.migrate_v1_to_v2(v1_data)
+        reconciler = (
+            v2_data["environments"]["prgs"]["services"]["gitea"]
+            ["identities"]["reconciler"]
+        )
+        self.assertEqual(reconciler["role"], "reconciler")
+        self.assertEqual(
+            reconciler["allowed_operations"],
+            migrate_profiles.RECONCILER_DEFAULT_ALLOWED,
+        )
+        self.assertEqual(
+            reconciler["forbidden_operations"],
+            migrate_profiles.RECONCILER_DEFAULT_FORBIDDEN,
+        )
+        for op in reconciler["allowed_operations"]:
+            self.assertEqual(gitea_config.normalize_operation(op), op)
+            self.assertTrue(op.startswith("gitea."))
+        assessment = reconciler_profile.assess_reconciler_profile(
+            reconciler["allowed_operations"],
+            reconciler["forbidden_operations"],
+        )
+        self.assertTrue(assessment["valid"])
+        self.assertNotIn(
+            "gitea.branch.delete", assessment["missing_recommended_operations"]
+        )
+
+    def test_reconciler_migration_idempotent_canonicalize(self):
+        """Second canonicalize of already-canonical ops is a no-op."""
+        first = migrate_profiles.canonicalize_operations(
+            list(migrate_profiles.RECONCILER_DEFAULT_ALLOWED)
+        )
+        second = migrate_profiles.canonicalize_operations(first)
+        self.assertEqual(first, second)
+        self.assertEqual(first, list(migrate_profiles.RECONCILER_DEFAULT_ALLOWED))
+
+    def test_reconciler_missing_required_fails_visibly(self):
+        """Missing gitea.pr.close after migration fails closed (not silent drop)."""
+        v1_data = {
+            "version": 1,
+            "profiles": {
+                "prgs-reconciler": {
+                    "base_url": "redacted-prgs-service",
+                    "username": "reconciler-agent",
+                    "auth": {"type": "keychain", "id": "reconciler-ref"},
+                    "execution_profile": "prgs-reconciler",
+                    "allowed_operations": ["read", "gitea.branch.delete"],
+                    "forbidden_operations": ["merge"],
+                }
+            },
+        }
+        with self.assertRaisesRegex(ValueError, "missing required"):
+            migrate_profiles.migrate_v1_to_v2(v1_data)
+
+    def test_unknown_operation_fails_visibly(self):
+        v1_data = {
+            "version": 1,
+            "profiles": {
+                "prgs-author": {
+                    "base_url": "redacted-prgs-service",
+                    "username": "jcwalker3",
+                    "auth": {"type": "keychain", "id": "hidden-author-ref"},
+                    "execution_profile": "prgs-author",
+                    "allowed_operations": ["read", "not.a.real.op"],
+                    "forbidden_operations": ["merge"],
+                }
+            },
+        }
+        with self.assertRaisesRegex(ValueError, "cannot be canonicalized"):
+            migrate_profiles.migrate_v1_to_v2(v1_data)
+
+    def test_role_inference_author_reviewer_merger_reconciler(self):
+        self.assertEqual(
+            migrate_profiles.infer_role("prgs-author", "prgs-author"), "author"
+        )
+        self.assertEqual(
+            migrate_profiles.infer_role("prgs-reviewer", "prgs-reviewer"),
+            "reviewer",
+        )
+        self.assertEqual(
+            migrate_profiles.infer_role("prgs-reconciler", "prgs-reconciler"),
+            "reconciler",
+        )
+        self.assertIsNone(
+            migrate_profiles.infer_role("prgs-merger", "prgs-merger")
+        )
+
 
 if __name__ == "__main__":
     unittest.main()
+

tests/test_reconciler_profile.py

@@ -82,6 +82,42 @@ class TestReconcilerProfileModel(unittest.TestCase):
             "reconciler",
         )
 
+    def test_branch_delete_is_recommended_for_reconciler(self):
+        self.assertIn(
+            "gitea.branch.delete",
+            reconciler_profile.RECONCILER_RECOMMENDED_OPERATIONS,
+        )
+        self.assertNotIn(
+            "gitea.branch.delete",
+            reconciler_profile.RECONCILER_REQUIRED_OPERATIONS,
+        )
+
+    def test_reconciler_with_branch_delete_stays_valid(self):
+        allowed = PRGS_RECONCILER_ALLOWED + ["gitea.branch.delete"]
+        result = reconciler_profile.assess_reconciler_profile(
+            allowed,
+            PRGS_RECONCILER_FORBIDDEN,
+        )
+        self.assertTrue(result["is_reconciler_profile"])
+        self.assertTrue(result["valid"])
+        self.assertNotIn(
+            "gitea.branch.delete", result["missing_recommended_operations"]
+        )
+        self.assertEqual(
+            mcp_server._role_kind(allowed, PRGS_RECONCILER_FORBIDDEN),
+            "reconciler",
+        )
+
+    def test_reconciler_without_branch_delete_reports_missing_recommended(self):
+        result = reconciler_profile.assess_reconciler_profile(
+            PRGS_RECONCILER_ALLOWED,
+            PRGS_RECONCILER_FORBIDDEN,
+        )
+        self.assertTrue(result["valid"])
+        self.assertIn(
+            "gitea.branch.delete", result["missing_recommended_operations"]
+        )
+
 
 if __name__ == "__main__":
     unittest.main()

tests/test_stale_review_decision_lock_cleanup.py

@@ -188,32 +188,20 @@ class TestActiveHardStopPreserved(unittest.TestCase):
         mcp_server._save_review_decision_lock(None)
         mcp_server.review_workflow_load.clear_review_workflow_load()
 
-    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_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_request_changes_still_blocks_same_pr(self):
-        _seed([RC_OPEN])  # request_changes on PR #701
+    def test_request_changes_still_blocks_everything(self):
+        _seed([RC_OPEN])
         for op in ("merge", "mark_ready", "review"):
             self.assertTrue(
-                mcp_server.terminal_review_hard_stop_reasons(701, op),
+                mcp_server.terminal_review_hard_stop_reasons(592, op),
                 msg=op,
             )
-        # Foreign PR review path is open (#693).
-        self.assertEqual(
-            mcp_server.terminal_review_hard_stop_reasons(592, "mark_ready"),
-            [],
-        )
 
 
 # --------------------------------------------------------------------------- #

tests/test_terminal_review_hard_stop.py

@@ -1,10 +1,10 @@
-"""Tests for the session hard-stop after a terminal review mutation (#332/#693).
+"""Tests for the session hard-stop after a terminal review mutation (#332).
 
 Extends the single-terminal-decision machinery (#211): after a live
-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.
+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.
 """
 import os
 import unittest
@@ -15,10 +15,7 @@ import mcp_server
 HEAD_SHA = "a" * 40
 
 
-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")
+def _lock(mutations=None, correction=False):
     return {
         "task": "review_pr",
         "remote": "prgs",
@@ -32,11 +29,9 @@ def _lock(mutations=None, correction=False, correction_pr=None):
         "ready_remote": None,
         "ready_org": None,
         "ready_repo": None,
-        "live_mutations": mutations,
+        "live_mutations": list(mutations or []),
         "correction_authorized": correction,
-        "correction_reason": "test" if correction else None,
-        "correction_pr_number": correction_pr if correction else None,
-        "correction_head_sha": None,
+        "correction_reason": None,
     }
 
 
@@ -69,35 +64,25 @@ class TestTerminalHardStopReasons(unittest.TestCase):
         self.assertEqual(
             mcp_server.terminal_review_hard_stop_reasons(5, "merge"), [])
 
-    def test_request_changes_blocks_same_pr_allows_other_pr_review(self):
+    def test_request_changes_blocks_everything(self):
         _seed([RC_A])
-        # Same PR: still hard-stopped for mark_ready/review/merge.
         for op in ("merge", "mark_ready", "review"):
-            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"))
+            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])
 
-    def test_approve_allows_same_pr_merge_and_other_pr_review(self):
+    def test_approve_allows_same_pr_merge_only(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"))
-        # #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"), [])
+        self.assertTrue(
+            mcp_server.terminal_review_hard_stop_reasons(6, "mark_ready"))
+        self.assertTrue(
+            mcp_server.terminal_review_hard_stop_reasons(6, "review"))
 
     def test_correction_reopens_review_path_only(self):
         _seed([RC_A], correction=True)
@@ -105,11 +90,9 @@ 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"), [])
-        # scoped correction never opens a cross-PR merge
+        # 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):
@@ -143,29 +126,14 @@ class TestMarkFinalHardStopWiring(unittest.TestCase):
         mcp_server._save_review_decision_lock(None)
         mcp_server.review_workflow_load.clear_review_workflow_load()
 
-    def test_mark_ready_same_pr_blocked_after_terminal_mutation(self):
+    def test_mark_ready_blocked_after_terminal_mutation(self):
         _seed([RC_A])
         result = mcp_server.gitea_mark_final_review_decision(
-            pr_number=5, action="approve", remote="prgs",
-            expected_head_sha=HEAD_SHA)
+            pr_number=6, action="approve", remote="prgs")
         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 {