fix: resolve conflicts for PR #511

Merge prgs/master into PR branch to restore mergeability.

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

gitea_mcp_server.py

@@ -211,6 +211,28 @@ def _effective_workspace_role() -> str:
     )
 
 
+def _actual_profile_role() -> str:
+    """Resolve the workspace role from the ACTIVE PROFILE alone (#540).
+
+    Unlike :func:`_effective_workspace_role`, this never consults
+    ``_preflight_resolved_role``. A task capability whose ``required_role_kind``
+    is ``author`` (e.g. ``comment_issue``) stamps ``_preflight_resolved_role =
+    "author"``; the #274/#475 role exemptions must key off the real profile
+    identity so that stamp cannot poison a genuine reviewer/merger/reconciler
+    session into being treated as an author. Author profiles still classify as
+    ``author`` here, so author blocking is preserved.
+    """
+    profile = get_profile()
+    role = _role_kind(
+        profile.get("allowed_operations") or [],
+        profile.get("forbidden_operations") or [],
+    )
+    return nwb.normalize_role_kind(
+        role,
+        profile_name=profile.get("profile_name"),
+    )
+
+
 def _resolve_preflight_workspace_path(worktree_path: str | None = None) -> str:
     """Resolve the namespace-scoped workspace root inspected by pre-flight guards."""
     role = _effective_workspace_role()
@@ -541,9 +563,19 @@ def _enforce_branches_only_author_mutation(worktree_path: str | None = None) ->
     Reviewer, merger, and reconciler roles are exempt: reconciler ``close_pr``
     is a Gitea metadata mutation and must not require ``GITEA_AUTHOR_WORKTREE``
     (#468). Non-author namespaces use dedicated workspace env vars (#510).
+
+    The exemption honours BOTH the effective workspace role and the actual
+    profile role (#540). ``comment_issue`` preflight stamps
+    ``_preflight_resolved_role = "author"`` (its ``required_role_kind``), which
+    would otherwise poison :func:`_effective_workspace_role` into classifying a
+    genuine reconciler as an author and defeat this exemption. Keying off the
+    real profile role as well preserves the exemption without weakening author
+    blocking — an actual author profile classifies as ``author`` in both.
     """
-    role = _effective_workspace_role()
-    if role in nwb.NON_AUTHOR_ROLES:
+    if (
+        _effective_workspace_role() in nwb.NON_AUTHOR_ROLES
+        or _actual_profile_role() in nwb.NON_AUTHOR_ROLES
+    ):
         return
     ctx = _resolve_namespace_mutation_context(worktree_path)
     workspace = ctx["workspace_path"]
@@ -651,6 +683,7 @@ def verify_preflight_purity(
                     f"{_format_preflight_files(reviewer_delta)}"
                 )
 
+    _enforce_root_checkout_guard(worktree_path)
     _enforce_branches_only_author_mutation(worktree_path)
     _clear_preflight_capability_state()
 
@@ -693,6 +726,28 @@ def _verify_role_mutation_workspace(
     verify_preflight_purity(remote, worktree_path=resolved, task=task)
     return resolved
 
+
+def _enforce_root_checkout_guard(worktree_path: str | None = None) -> None:
+    """#475: fail closed when the stable control checkout is contaminated."""
+    ctx = _resolve_author_mutation_context(worktree_path)
+    canonical_root = ctx["canonical_repo_root"]
+    workspace = ctx["workspace_path"]
+    git_state = issue_lock_worktree.read_worktree_git_state(canonical_root)
+    remote_master_sha = root_checkout_guard.resolve_remote_master_sha(canonical_root)
+    assessment = root_checkout_guard.assess_root_checkout_guard(
+        workspace_path=workspace,
+        canonical_repo_root=canonical_root,
+        current_branch=git_state.get("current_branch"),
+        head_sha=git_state.get("head_sha"),
+        porcelain_status=git_state.get("porcelain_status") or "",
+        remote_master_sha=remote_master_sha,
+        resolved_role=_preflight_resolved_role,
+        actual_role=_actual_profile_role(),
+    )
+    if assessment["block"]:
+        raise RuntimeError(root_checkout_guard.format_root_checkout_guard_error(assessment))
+
+
 from mcp.server.fastmcp import FastMCP  # noqa: E402
 
 from gitea_auth import (  # noqa: E402
@@ -714,6 +769,7 @@ import role_session_router  # noqa: E402
 import role_namespace_gate  # noqa: E402
 import task_capability_map  # noqa: E402
 import review_proofs  # noqa: E402
+import review_workflow_boundary  # noqa: E402
 import review_workflow_load  # noqa: E402
 import agent_temp_artifacts
 import issue_lock_worktree  # noqa: E402
@@ -724,10 +780,14 @@ import stacked_pr_support  # noqa: E402
 import merge_approval_gate  # noqa: E402
 import already_landed_reconcile  # noqa: E402
 import author_mutation_worktree  # noqa: E402
+import root_checkout_guard  # noqa: E402
+import remote_repo_guard  # noqa: E402
 import issue_claim_heartbeat  # noqa: E402
 import issue_work_duplicate_gate  # noqa: E402
 import reviewer_pr_lease  # noqa: E402
+import merger_lease_adoption  # noqa: E402
 import merged_cleanup_reconcile  # noqa: E402
+import worktree_cleanup_audit  # noqa: E402
 import reconciler_profile  # noqa: E402
 import reconciliation_workflow  # noqa: E402
 import audit_reconciliation_mode  # noqa: E402
@@ -735,6 +795,7 @@ import review_merge_state_machine  # noqa: E402
 import pr_work_lease  # noqa: E402
 import native_mcp_preference  # noqa: E402
 import thread_state_ledger_validator  # noqa: E402
+import worktree_cleanup_audit  # noqa: E402
 
 
 # Keyed issue-lock storage (#443): per remote/org/repo/issue files under
@@ -1136,11 +1197,49 @@ def _resolve(remote: str, host: str | None, org: str | None, repo: str | None):
     if remote not in REMOTES:
         raise ValueError(f"Unknown remote '{remote}'. Choose from: {list(REMOTES)}")
     profile = REMOTES[remote]
-    return (
-        host or profile["host"],
-        org or profile["org"],
-        repo or profile["repo"],
+    resolved_host = host or profile["host"]
+    resolved_org = org or profile["org"]
+    resolved_repo = repo or profile["repo"]
+    _enforce_remote_repo_guard(
+        remote,
+        resolved_org,
+        resolved_repo,
+        org_explicit=org is not None,
+        repo_explicit=repo is not None,
     )
+    return (resolved_host, resolved_org, resolved_repo)
+
+
+def _enforce_remote_repo_guard(
+    remote: str,
+    resolved_org: str,
+    resolved_repo: str,
+    *,
+    org_explicit: bool,
+    repo_explicit: bool,
+) -> None:
+    """Fail closed on a remote/repo mismatch vs. the local git remote (#530).
+
+    Best-effort: bypassed under pytest unless ``GITEA_FORCE_REMOTE_REPO_CHECK`` is
+    set, so the unit suite (which calls tools with bare remotes against mocked APIs)
+    is unaffected. In production it protects every read/lookup/mutation tool because
+    they all resolve targets through :func:`_resolve`.
+    """
+    if "pytest" in sys.modules and not os.environ.get(
+        "GITEA_FORCE_REMOTE_REPO_CHECK"
+    ):
+        return
+    local_remote_url = _local_git_remote_url(remote)
+    assessment = remote_repo_guard.assess_remote_repo_match(
+        remote=remote,
+        resolved_org=resolved_org,
+        resolved_repo=resolved_repo,
+        local_remote_url=local_remote_url,
+        org_explicit=org_explicit,
+        repo_explicit=repo_explicit,
+    )
+    if assessment["block"]:
+        raise RuntimeError(remote_repo_guard.format_remote_repo_guard_error(assessment))
 
 
 def _auth(host: str) -> str:
@@ -2383,10 +2482,18 @@ def _review_decision_session_reasons(lock: dict | None) -> list[str]:
     return reasons
 
 
-def init_review_decision_lock(remote: str | None, task: str | None):
+def init_review_decision_lock(remote: str | None, task: str | None, force: bool = True):
     """Seed read-only-until-ready state for reviewer PR review tasks."""
     if task != "review_pr":
         return
+    if not force:
+        lock = _load_review_decision_lock()
+        if lock is not None:
+            if lock.get("remote") == remote and lock.get("session_pid") == os.getpid():
+                env_lock = (os.environ.get(SESSION_PROFILE_LOCK_ENV) or "").strip()
+                stored_lock = (lock.get("session_profile_lock") or "").strip()
+                if not env_lock or not stored_lock or env_lock == stored_lock:
+                    return
     review_workflow_load.clear_review_workflow_load()
     profile = get_profile()
     profile_name = (profile.get("profile_name") or "").strip()
@@ -2779,6 +2886,85 @@ def gitea_get_pr_review_feedback(
     }
 
 
+def _fetch_pr_lease_comments_safe(
+    pr_number: int,
+    *,
+    remote: str,
+    host: str | None,
+    org: str | None,
+    repo: str | None,
+    limit: int = 100,
+    require_open: bool = False,
+) -> dict:
+    """Fetch PR thread comments with structured fail-closed errors (#519)."""
+    h, o, r = _resolve(remote, host, org, repo)
+    auth = _auth(h)
+    resolved_repo = f"{o}/{r}"
+    pr_url = f"{repo_api_url(h, o, r)}/pulls/{pr_number}"
+    try:
+        pr = api_request("GET", pr_url, auth)
+    except RuntimeError as exc:
+        return {
+            "success": False,
+            "comments": [],
+            "reasons": [
+                "PR lookup failed before conflict-fix push assessment "
+                f"(pr_number={pr_number}, repo={resolved_repo}, remote={remote}): "
+                f"{_redact(str(exc))}"
+            ],
+            "pr_lookup": "failed",
+            "resolved_repo": resolved_repo,
+            "remote": remote,
+            "pr_number": pr_number,
+        }
+    pr_state = (pr.get("state") or "").strip().lower()
+    if require_open and pr_state != "open":
+        return {
+            "success": False,
+            "comments": [],
+            "reasons": [
+                f"PR #{pr_number} on {resolved_repo} is not open "
+                f"(state={pr_state or 'unknown'})"
+            ],
+            "pr_lookup": "not_open",
+            "resolved_repo": resolved_repo,
+            "remote": remote,
+            "pr_number": pr_number,
+            "head_sha": (pr.get("head") or {}).get("sha"),
+        }
+    api = f"{repo_api_url(h, o, r)}/issues/{pr_number}/comments"
+    try:
+        comments = api_request("GET", api, auth)
+    except RuntimeError as exc:
+        return {
+            "success": False,
+            "comments": [],
+            "reasons": [
+                "PR comment fetch failed during conflict-fix push assessment "
+                f"(pr_number={pr_number}, issue_index={pr_number}, "
+                f"repo={resolved_repo}, remote={remote}): "
+                f"{_redact(str(exc))}"
+            ],
+            "pr_lookup": "ok",
+            "resolved_repo": resolved_repo,
+            "remote": remote,
+            "pr_number": pr_number,
+            "head_sha": (pr.get("head") or {}).get("sha"),
+        }
+    if not isinstance(comments, list):
+        comments = []
+    return {
+        "success": True,
+        "comments": list(comments[:limit]),
+        "reasons": [],
+        "pr_lookup": "ok",
+        "resolved_repo": resolved_repo,
+        "remote": remote,
+        "pr_number": pr_number,
+        "head_sha": (pr.get("head") or {}).get("sha"),
+    }
+
+
 def _list_pr_lease_comments(
     pr_number: int,
     *,
@@ -2788,7 +2974,15 @@ def _list_pr_lease_comments(
     repo: str | None,
     limit: int = 100,
 ) -> list[dict]:
-    """Fetch PR/issue thread comments used for reviewer/conflict-fix leases."""
+    """Fetch PR/issue thread comments used for reviewer/conflict-fix leases.
+
+    Intentionally does **not** pre-fetch the PR via GET /pulls/{n}. Shared
+    callers (merge approval feedback, reviewer lease gates) depend on a single
+    comments GET so mock sequences and fail-open #485 non-list handling stay
+    stable. Structured PR-lookup failures belong only to
+    :func:`_fetch_pr_lease_comments_safe` used by conflict-fix push assessment
+    (#519).
+    """
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
     api = f"{repo_api_url(h, o, r)}/issues/{pr_number}/comments"
@@ -4303,6 +4497,59 @@ def _remote_branch_exists(h: str, o: str, r: str, auth: str, branch: str) -> boo
         raise
 
 
+@mcp.tool()
+def gitea_capture_branches_worktree_snapshot(
+    open_pr_branches: list[str] | None = None,
+    active_lock_branch: str | None = None,
+    leased_paths: list[str] | None = None,
+    worktree_path: str | None = None,
+) -> dict:
+    """Read-only: capture ``branches/`` and worktree audit snapshot (#404)."""
+    read_block = _profile_operation_gate("gitea.read")
+    if read_block:
+        return {
+            "success": False,
+            "reasons": read_block,
+            "permission_report": _permission_block_report("gitea.read"),
+        }
+    root = PROJECT_ROOT
+    if worktree_path:
+        root = os.path.realpath(os.path.abspath(worktree_path))
+        git_root = _get_git_root(root)
+        if git_root:
+            root = git_root
+    snapshot = worktree_cleanup_audit.capture_branches_worktree_snapshot(
+        root,
+        open_pr_branches=open_pr_branches,
+        active_lock_branch=active_lock_branch,
+        leased_paths=leased_paths,
+    )
+    return {"success": True, "snapshot": snapshot}
+
+
+@mcp.tool()
+def gitea_assess_worktree_cleanup_integrity(
+    before_snapshot: dict,
+    after_snapshot: dict,
+    removals: list[dict] | None = None,
+    explained_missing: dict[str, str] | None = None,
+) -> dict:
+    """Read-only: reconcile cleanup before/after snapshots (#404)."""
+    read_block = _profile_operation_gate("gitea.read")
+    if read_block:
+        return {
+            "integrity_passed": False,
+            "reasons": read_block,
+            "permission_report": _permission_block_report("gitea.read"),
+        }
+    return worktree_cleanup_audit.assess_worktree_cleanup_integrity(
+        before=before_snapshot,
+        after=after_snapshot,
+        removals=removals,
+        explained_missing=explained_missing,
+    )
+
+
 @mcp.tool()
 def gitea_reconcile_merged_cleanups(
     dry_run: bool = True,
@@ -4654,6 +4901,90 @@ def gitea_scan_already_landed_open_prs(
     }
 
 
+@mcp.tool()
+def gitea_audit_worktree_cleanup(
+    remote: str = "dadeschools",
+    host: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+    ttl_hours: float = worktree_cleanup_audit.DEFAULT_TTL_HOURS,
+) -> dict:
+    """Read-only: classify every session-owned worktree under ``branches/`` (#401).
+
+    Audits the local ``branches/`` directory, classifying each worktree as
+    active open PR, active issue work, dirty local, clean stale removable,
+    detached review leftover, or unsafe/unknown. Open PR branch heads are
+    fetched live so a worktree tied to an open PR is never marked removable;
+    the active issue-lock branch is read from the local lock file and treated
+    as active work. Deletes nothing and mutates no Gitea state.
+
+    Fails closed if the live open-PR list cannot be fetched: without it,
+    removability cannot be proven, so no candidates are returned.
+
+    Args:
+        remote: Known instance — 'dadeschools' or 'prgs'.
+        host: Override the Gitea host.
+        org: Override the owner/organization.
+        repo: Override the repository name.
+        ttl_hours: Age (hours) after which a clean issue/conflict-fix
+            worktree becomes stale-removable (default from
+            GITEA_WORKTREE_TTL_HOURS).
+
+    Returns:
+        dict with per-worktree classifications, counts, removable
+        candidates, and the ``git worktree list`` verification proof.
+    """
+    read_block = _profile_operation_gate("gitea.read")
+    if read_block:
+        return {
+            "success": False,
+            "performed": False,
+            "reasons": read_block,
+            "permission_report": _permission_block_report("gitea.read"),
+        }
+
+    try:
+        h, o, r = _resolve(remote, host, org, repo)
+        auth = _auth(h)
+        open_prs = api_get_all(f"{repo_api_url(h, o, r)}/pulls?state=open", auth)
+    except Exception as exc:
+        return {
+            "success": False,
+            "performed": False,
+            "open_pr_state_verified": False,
+            "reasons": [
+                "could not fetch live open PRs; removability unverified "
+                f"(fail closed): {_redact(str(exc))}"
+            ],
+        }
+
+    open_pr_branches = {
+        str((pr.get("head") or {}).get("ref"))
+        for pr in open_prs
+        if (pr.get("head") or {}).get("ref")
+    }
+
+    active_issue_branches: set[str] = set()
+    lock = merged_cleanup_reconcile.read_issue_lock(ISSUE_LOCK_FILE)
+    if lock and lock.get("branch_name"):
+        active_issue_branches.add(str(lock["branch_name"]).strip())
+
+    report = worktree_cleanup_audit.audit_branches_directory(
+        PROJECT_ROOT,
+        open_pr_branches=open_pr_branches,
+        active_issue_branches=active_issue_branches,
+        now=datetime.now(timezone.utc),
+        ttl_hours=ttl_hours,
+    )
+    return {
+        "success": True,
+        "performed": False,
+        "open_pr_state_verified": True,
+        "task_mode": "work-issue",
+        **report,
+    }
+
+
 @mcp.tool()
 def gitea_reconcile_already_landed_pr(
     pr_number: int,
@@ -5351,7 +5682,10 @@ def gitea_acquire_reviewer_pr_lease(
         "target_branch_sha": target_branch_sha,
         "repo": repo_label,
         "comment_id": posted.get("id"),
-    })
+    }, lease_provenance=merger_lease_adoption.build_lease_provenance(
+        source=merger_lease_adoption.SOURCE_ACQUIRE,
+        comment_id=posted.get("id"),
+    ))
     return {
         "success": True,
         "acquired": True,
@@ -5363,6 +5697,175 @@ def gitea_acquire_reviewer_pr_lease(
     }
 
 
+@mcp.tool()
+def gitea_adopt_merger_pr_lease(
+    pr_number: int,
+    worktree: str,
+    expected_head_sha: str,
+    issue_number: int | None = None,
+    session_id: str | None = None,
+    remote: str = "dadeschools",
+    host: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+) -> dict:
+    """Adopt an active reviewer PR lease for a merger session (#536).
+
+    Merger sessions must not manually seed ``reviewer_pr_lease._SESSION_LEASE``.
+    This tool posts durable adoption proof on the PR thread and records
+    sanctioned in-session lease provenance so ``gitea_merge_pr`` can proceed
+    after a separate reviewer session held the lease.
+
+    Args:
+        pr_number: Open PR to adopt.
+        worktree: Merger workspace under ``branches/``.
+        expected_head_sha: Approved head the merger will pin during merge.
+        issue_number: Optional linked issue for the adoption record.
+        session_id: Optional merger session id (generated when omitted).
+        remote: Known instance — 'dadeschools' or 'prgs'.
+        host: Override the Gitea host.
+        org: Override the owner/organization.
+        repo: Override the repository name.
+
+    Returns:
+        dict with adoption proof, session lease, and block reasons when the
+        guarded preconditions are not met.
+    """
+    read_block = _profile_operation_gate("gitea.read")
+    if read_block:
+        return {
+            "success": False,
+            "adopted": False,
+            "reasons": read_block,
+            "permission_report": _permission_block_report("gitea.read"),
+        }
+    comment_block = _profile_operation_gate("gitea.pr.comment")
+    if comment_block:
+        return {
+            "success": False,
+            "adopted": False,
+            "reasons": comment_block,
+            "permission_report": _permission_block_report("gitea.pr.comment"),
+        }
+    merge_block = _profile_operation_gate("gitea.pr.merge")
+    if merge_block:
+        return {
+            "success": False,
+            "adopted": False,
+            "reasons": merge_block,
+            "permission_report": _permission_block_report("gitea.pr.merge"),
+        }
+
+    _verify_role_mutation_workspace(
+        remote, worktree=worktree, task="adopt_merger_pr_lease"
+    )
+    h, o, r = _resolve(remote, host, org, repo)
+    auth = _auth(h)
+    profile = get_profile()
+    profile_name = profile.get("profile_name") or "unknown"
+    identity = _authenticated_username(remote) or profile.get("username") or ""
+    sid = (session_id or "").strip() or reviewer_pr_lease.new_session_id()
+    repo_label = f"{o}/{r}"
+
+    pr_live = api_request(
+        "GET", f"{repo_api_url(h, o, r)}/pulls/{pr_number}", auth) or {}
+    pr_state = (pr_live.get("state") or "").strip().lower()
+    pr_open = pr_state == "open" and not pr_live.get("merged")
+    live_head = (pr_live.get("head", {}) or {}).get("sha") or pr_live.get(
+        "head_sha"
+    )
+
+    feedback = gitea_get_pr_review_feedback(
+        pr_number=pr_number,
+        remote=remote,
+        host=host,
+        org=org,
+        repo=repo,
+    )
+    approval_at_head = bool(feedback.get("approval_at_current_head"))
+
+    comments = _fetch_pr_comments(
+        pr_number, remote=remote, host=host, org=org, repo=repo)
+    assessment = merger_lease_adoption.assess_adopt_merger_lease(
+        comments,
+        pr_number=pr_number,
+        adopter_identity=identity,
+        adopter_profile=profile_name,
+        adopter_session_id=sid,
+        repo=repo_label,
+        issue_number=issue_number,
+        worktree=worktree,
+        expected_head_sha=expected_head_sha,
+        live_head_sha=live_head,
+        approval_at_head=approval_at_head,
+        pr_open=pr_open,
+    )
+    if not assessment.get("adopt_allowed"):
+        return {
+            "success": False,
+            "adopted": False,
+            "pr_number": pr_number,
+            "reasons": assessment.get("reasons") or [],
+            "active_lease": assessment.get("active_lease"),
+            "expected_head_sha": expected_head_sha,
+            "live_head_sha": live_head,
+        }
+
+    active = assessment.get("active_lease") or {}
+    body = assessment["adoption_body"]
+    comment_url = f"{repo_api_url(h, o, r)}/issues/{pr_number}/comments"
+    with _audited(
+        "comment_pr",
+        host=h,
+        remote=remote,
+        org=o,
+        repo=r,
+        pr_number=pr_number,
+        request_metadata={"source": "adopt_merger_pr_lease"},
+    ):
+        posted = api_request("POST", comment_url, auth, {"body": body})
+
+    provenance = merger_lease_adoption.build_lease_provenance(
+        source=merger_lease_adoption.SOURCE_ADOPT,
+        comment_id=posted.get("id"),
+        adopted_from_session_id=active.get("session_id"),
+        adopted_from_profile=active.get("profile"),
+        adopted_from_reviewer_identity=active.get("reviewer_identity"),
+        adoption_reason=merger_lease_adoption.DEFAULT_ADOPTION_REASON,
+    )
+    session_lease = reviewer_pr_lease.record_session_lease({
+        "pr_number": pr_number,
+        "issue_number": issue_number or active.get("issue_number"),
+        "session_id": sid,
+        "reviewer_identity": identity,
+        "profile": profile_name,
+        "worktree": worktree,
+        "phase": "adopted",
+        "candidate_head": live_head,
+        "target_branch": active.get("target_branch") or "master",
+        "target_branch_sha": active.get("target_branch_sha"),
+        "repo": repo_label,
+        "comment_id": posted.get("id"),
+    }, lease_provenance=provenance)
+
+    return {
+        "success": True,
+        "adopted": True,
+        "pr_number": pr_number,
+        "session_id": sid,
+        "adoption_comment_id": posted.get("id"),
+        "adopted_from_session_id": active.get("session_id"),
+        "adopted_from_profile": active.get("profile"),
+        "adopted_from_reviewer_identity": active.get("reviewer_identity"),
+        "adoption_reason": merger_lease_adoption.DEFAULT_ADOPTION_REASON,
+        "expected_head_sha": expected_head_sha,
+        "live_head_sha": live_head,
+        "session_lease": session_lease,
+        "lease_provenance": provenance,
+        "reasons": [],
+    }
+
+
 @mcp.tool()
 def gitea_heartbeat_reviewer_pr_lease(
     pr_number: int,
@@ -5423,6 +5926,17 @@ def gitea_heartbeat_reviewer_pr_lease(
     ):
         posted = api_request("POST", comment_url, auth, {"body": body})
 
+    prior_provenance = (session.get("lease_provenance") or {}).copy()
+    heartbeat_provenance = merger_lease_adoption.build_lease_provenance(
+        source=merger_lease_adoption.SOURCE_HEARTBEAT,
+        comment_id=posted.get("id"),
+        adopted_from_session_id=prior_provenance.get("adopted_from_session_id"),
+        adopted_from_profile=prior_provenance.get("adopted_from_profile"),
+        adopted_from_reviewer_identity=prior_provenance.get(
+            "adopted_from_reviewer_identity"
+        ),
+        adoption_reason=prior_provenance.get("adoption_reason"),
+    )
     updated = reviewer_pr_lease.record_session_lease({
         **session,
         "phase": phase,
@@ -5430,7 +5944,8 @@ def gitea_heartbeat_reviewer_pr_lease(
         "candidate_head": candidate_head or session.get("candidate_head"),
         "target_branch_sha": target_branch_sha or session.get("target_branch_sha"),
         "last_comment_id": posted.get("id"),
-    })
+        "comment_id": posted.get("id"),
+    }, lease_provenance=heartbeat_provenance)
     return {
         "success": True,
         "posted": True,
@@ -5591,6 +6106,118 @@ def gitea_cleanup_post_merge_moot_lease(
     return report
 
 
+@mcp.tool()
+def gitea_release_reviewer_pr_lease(
+    pr_number: int,
+    worktree: str | None = None,
+    remote: str = "dadeschools",
+    host: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+) -> dict:
+    """Safely release an active reviewer lease owned by the current session on an open PR.
+
+    This provides a canonical way to clean up reviewer leases when a review fails
+    before submission.
+
+    Args:
+        pr_number: The PR number whose reviewer lease to release.
+        worktree: Optional worktree path (resolves automatically if not supplied).
+        remote: Known instance — 'dadeschools' or 'prgs'.
+        host: Override the Gitea host.
+        org: Override the owner/organization.
+        repo: Override the repository name.
+
+    Returns:
+        dict reporting release status.
+    """
+    _verify_role_mutation_workspace(
+        remote, worktree=worktree, task="review_pr"
+    )
+    h, o, r = _resolve(remote, host, org, repo)
+    auth = _auth(h)
+
+    comments = _fetch_pr_comments(
+        pr_number, remote=remote, host=host, org=org, repo=repo
+    )
+    active = reviewer_pr_lease.find_active_reviewer_lease(
+        comments, pr_number=pr_number
+    )
+
+    if not active:
+        return {
+            "success": True,
+            "released": False,
+            "reasons": ["no active reviewer lease found on PR"],
+        }
+
+    identity = _authenticated_username(remote) or ""
+    session = reviewer_pr_lease.get_session_lease() or {}
+    session_id = session.get("session_id")
+
+    owner_identity = active.get("reviewer_identity")
+    owner_session_id = active.get("session_id")
+
+    # Authorize release: identity matches, session_id matches, or lease is expired/reclaimable
+    authorized = False
+    reasons = []
+    if owner_identity and identity and owner_identity == identity:
+        authorized = True
+    elif owner_session_id and session_id and owner_session_id == session_id:
+        authorized = True
+    else:
+        freshness = reviewer_pr_lease.classify_lease_freshness(active)
+        if freshness in {"expired", "reclaimable"}:
+            authorized = True
+
+    if not authorized:
+        return {
+            "success": False,
+            "released": False,
+            "reasons": [
+                f"unauthorized to release active lease: owned by {owner_identity} "
+                f"(session {owner_session_id})"
+            ],
+        }
+
+    # Format and post release comment
+    body = reviewer_pr_lease.format_lease_body(
+        repo=active.get("repo") or f"{o}/{r}",
+        pr_number=pr_number,
+        issue_number=active.get("issue_number"),
+        reviewer_identity=owner_identity or identity,
+        profile=active.get("profile") or "reviewer",
+        session_id=owner_session_id or session_id or "unknown",
+        worktree=active.get("worktree") or "",
+        phase="released",
+        candidate_head=active.get("candidate_head"),
+        target_branch=active.get("target_branch") or "master",
+        target_branch_sha=active.get("target_branch_sha"),
+        blocker="manual-release",
+    )
+
+    comment_url = f"{repo_api_url(h, o, r)}/issues/{pr_number}/comments"
+    with _audited(
+        "comment_pr",
+        host=h,
+        remote=remote,
+        org=o,
+        repo=r,
+        pr_number=pr_number,
+        request_metadata={"source": "release_reviewer_pr_lease"},
+    ):
+        posted = api_request("POST", comment_url, auth, {"body": body})
+
+    reviewer_pr_lease.clear_session_lease()
+
+    return {
+        "success": True,
+        "released": True,
+        "comment_id": posted.get("id"),
+        "reasons": [],
+    }
+
+
 @mcp.tool()
 def gitea_list_issue_comments(
     issue_number: int,
@@ -5679,6 +6306,7 @@ def gitea_create_issue_comment(
     host: str | None = None,
     org: str | None = None,
     repo: str | None = None,
+    worktree_path: str | None = None,
 ) -> dict:
     """Post a markdown comment to a Gitea issue's discussion thread.
 
@@ -5700,6 +6328,7 @@ def gitea_create_issue_comment(
         host: Override the Gitea host.
         org: Override the owner/organization.
         repo: Override the repository name.
+        worktree_path: Optional path to verify branches-only guard.
 
     Returns:
         dict with 'success', 'comment_id', and 'issue_number' ('url' only
@@ -5708,7 +6337,7 @@ def gitea_create_issue_comment(
         (permission blocks also carry a structured 'permission_report',
         #142).
     """
-    verify_preflight_purity(remote, task="comment_issue")
+    verify_preflight_purity(remote, worktree_path=worktree_path, task="comment_issue")
     gate_reasons = _profile_operation_gate("gitea.issue.comment")
     reasons = list(gate_reasons)
     if not (body or "").strip():
@@ -6944,14 +7573,42 @@ def gitea_get_runtime_context(
     return result
 
 
+@mcp.tool()
+def gitea_record_pre_review_command(
+    command: str,
+    cwd: str | None = None,
+    classification: str | None = None,
+) -> dict:
+    """Classify and record a command executed before workflow load (#403).
+
+    Read-only with respect to Gitea API. Pre-review inventory/diagnostic commands
+    may be recorded as allowed; boundary violations block reviewer mutations.
+    """
+    recorded = review_workflow_boundary.record_pre_review_command(
+        command,
+        cwd=cwd,
+        project_root=PROJECT_ROOT,
+        classification=classification,
+    )
+    boundary_state = review_workflow_boundary.assess_boundary_status(PROJECT_ROOT)
+    return {
+        "success": True,
+        "recorded": recorded,
+        "boundary_status": boundary_state.get("boundary_status"),
+        "boundary_clean": boundary_state.get("boundary_clean"),
+        "reasons": list(boundary_state.get("reasons") or []),
+    }
+
+
 @mcp.tool()
 def gitea_load_review_workflow(
     prompt_text: str | None = None,
 ) -> dict:
-    """Load and record canonical review-merge workflow proof for this session (#389).
+    """Load and record canonical review-merge workflow proof for this session (#389, #403).
 
     Read-only with respect to Gitea API; records in-process workflow source/hash
-    proof required before reviewer review or merge mutations.
+    proof and session boundary state required before reviewer review or merge
+    mutations.
     """
     try:
         recorded = review_workflow_load.record_review_workflow_load(
@@ -6963,8 +7620,11 @@ def gitea_load_review_workflow(
             "reasons": [str(exc)],
             "recovery_handoff": review_workflow_load.recovery_handoff_without_replay(),
         }
+    boundary_reasons = review_workflow_boundary.boundary_blockers(PROJECT_ROOT)
+    helper = review_workflow_boundary.workflow_load_helper_result(
+        recorded, PROJECT_ROOT)
     return {
-        "success": True,
+        "success": not boundary_reasons,
         "loaded": True,
         "workflow_source": recorded["workflow_source"],
         "task_mode": recorded["task_mode"],
@@ -6976,7 +7636,14 @@ def gitea_load_review_workflow(
             "prompt_conflicts_with_workflow"],
         "prompt_conflict_reasons": recorded.get("prompt_conflict_reasons") or [],
         "workflow_load_proof_present": True,
-        "reasons": [],
+        "boundary_status": recorded.get("boundary_status"),
+        "boundary_clean": recorded.get("boundary_clean"),
+        "workflow_load_helper_result": helper,
+        "reasons": boundary_reasons,
+        "recovery_handoff": (
+            review_workflow_load.recovery_handoff_without_replay()
+            if boundary_reasons else []
+        ),
     }
 
 
@@ -7471,22 +8138,39 @@ def gitea_assess_conflict_fix_push(
             "reasons": read_block,
             "permission_report": _permission_block_report("gitea.read"),
         }
-    comments = _list_pr_lease_comments(
+    fetched = _fetch_pr_lease_comments_safe(
         pr_number,
         remote=remote,
         host=host,
         org=org,
         repo=repo,
+        require_open=True,
     )
-    return pr_work_lease.assess_conflict_fix_push(
+    if not fetched["success"]:
+        return {
+            "push_allowed": False,
+            "block": True,
+            "reasons": fetched["reasons"],
+            "pr_lookup": fetched.get("pr_lookup"),
+            "resolved_repo": fetched.get("resolved_repo"),
+            "remote": fetched.get("remote"),
+            "pr_number": pr_number,
+            "assessment_failed": True,
+        }
+    assessment = pr_work_lease.assess_conflict_fix_push(
         pr_number=pr_number,
-        comments=comments,
+        comments=fetched["comments"],
         branch_head_before=branch_head_before,
         branch_head_after=branch_head_after,
         worktree_path=worktree_path,
         push_cwd=push_cwd,
         is_fast_forward=is_fast_forward,
     )
+    assessment["pr_lookup"] = fetched.get("pr_lookup")
+    assessment["resolved_repo"] = fetched.get("resolved_repo")
+    assessment["remote"] = fetched.get("remote")
+    assessment["live_pr_head_sha"] = fetched.get("head_sha")
+    return assessment
 
 
 @mcp.tool()
@@ -8124,6 +8808,7 @@ def gitea_resolve_task_capability(
         init_review_decision_lock(
             remote if remote in REMOTES else None,
             task,
+            force=False,
         )
 
     record_mutation_authority(profile["profile_name"], username, remote if remote in REMOTES else None, task)