diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py
index b64d637..6e8aed9 100644
--- a/gitea_mcp_server.py
+++ b/gitea_mcp_server.py
@@ -1282,12 +1282,18 @@ def _verify_role_mutation_workspace(
     worktree_path: str | None = None,
     worktree: str | None = None,
     task: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
 ) -> str:
     """Bind reviewer/merger mutations to the active namespace workspace (#510).
 
     #683: must NOT early-return solely because pytest/unittest is loaded.
     Production workspace binding always runs; test isolation uses explicit
     env fixtures / force-on flags, never a production short-circuit here.
+
+    org/repo are forwarded into anti-stomp (#604) so callers that pass explicit
+    repository targets (e.g. prgs + Gitea-Tools when REMOTES defaults to
+    Timesheet) do not fail closed with wrong_repo after a successful resolve.
     """
 
     # Check running runtimes to prevent stale mutations
@@ -1350,7 +1356,13 @@ def _verify_role_mutation_workspace(
             )
         )
     resolved = assessment["mutation_workspace"]
-    verify_preflight_purity(remote, worktree_path=resolved, task=task)
+    verify_preflight_purity(
+        remote,
+        worktree_path=resolved,
+        task=task,
+        org=org,
+        repo=repo,
+    )
     return resolved
 
 
@@ -4522,7 +4534,11 @@ def _evaluate_pr_review_submission(
 ) -> dict:
     """Shared gate chain for live submit and dry-run review tools."""
     _verify_role_mutation_workspace(
-        remote, worktree_path=worktree_path, task="review_pr"
+        remote,
+        worktree_path=worktree_path,
+        task="review_pr",
+        org=org,
+        repo=repo,
     )
     action = (action or "").strip().lower()
     workflow_blockers = _review_workflow_load_gate_reasons() if live else []
@@ -7620,7 +7636,11 @@ def gitea_merge_pr(
         available. Never secrets.
     """
     _verify_role_mutation_workspace(
-        remote, worktree_path=worktree_path, task="merge_pr"
+        remote,
+        worktree_path=worktree_path,
+        task="merge_pr",
+        org=org,
+        repo=repo,
     )
     allowed = get_profile()["allowed_operations"]
     forbidden = get_profile()["forbidden_operations"]
@@ -10653,8 +10673,14 @@ def gitea_acquire_reviewer_pr_lease(
 
     # task=acquire_reviewer_pr_lease so verify_preflight_purity runs shared #604
     # anti-stomp for the declared lease-acquire mutation inventory entry.
+    # Forward explicit org/repo so anti-stomp does not default bare remote=prgs
+    # to Timesheet when the local git remote is Gitea-Tools (#604 wrong_repo).
     _verify_role_mutation_workspace(
-        remote, worktree=worktree, task="acquire_reviewer_pr_lease"
+        remote,
+        worktree=worktree,
+        task="acquire_reviewer_pr_lease",
+        org=org,
+        repo=repo,
     )
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
@@ -10737,6 +10763,200 @@ def gitea_acquire_reviewer_pr_lease(
     }
 
 
+@mcp.tool()
+def gitea_acquire_merger_pr_lease(
+    pr_number: int,
+    worktree: str,
+    candidate_head: str | None = None,
+    target_branch: str = "master",
+    target_branch_sha: str | None = None,
+    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:
+    """Acquire a per-PR lease for a merger session when no reviewer lease exists (#718).
+
+    Merger sessions should normally adopt an active reviewer lease using
+    gitea_adopt_merger_pr_lease. However, if no active reviewer lease exists
+    or it has expired, a merger can acquire one directly using this tool.
+    """
+    read_block = _profile_operation_gate("gitea.read")
+    if read_block:
+        return {
+            "success": False,
+            "acquired": 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,
+            "acquired": 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,
+            "acquired": False,
+            "reasons": merge_block,
+            "permission_report": _permission_block_report("gitea.pr.merge"),
+        }
+
+    head_pin = (candidate_head or "").strip()
+    if not head_pin:
+        return {
+            "success": False,
+            "acquired": False,
+            "reasons": [
+                "candidate_head is required for merger lease acquisition "
+                "(exact-head scoping; fail closed)"
+            ],
+        }
+
+    try:
+        _verify_role_mutation_workspace(
+            remote,
+            worktree=worktree,
+            task="acquire_merger_pr_lease",
+            org=org,
+            repo=repo,
+        )
+    except RuntimeError as e:
+        return {
+            "success": False,
+            "acquired": False,
+            "reasons": [str(e)],
+        }
+
+    h, o, r = _resolve(remote, host, org, repo)
+    auth = _auth(h)
+    profile = get_profile()
+    identity = _authenticated_username(h) or profile.get("username") or ""
+    sid = (session_id or "").strip() or reviewer_pr_lease.new_session_id()
+    repo_label = f"{o}/{r}"
+
+    comments = _fetch_pr_comments(
+        pr_number, remote=remote, host=host, org=org, repo=repo)
+
+    pr_live = api_request(
+        "GET", f"{repo_api_url(h, o, r)}/pulls/{pr_number}", auth) or {}
+    live_head = (
+        (pr_live.get("head") or {}).get("sha")
+        or pr_live.get("head_sha")
+        or ""
+    )
+    live_head = str(live_head).strip()
+    # Fail closed when live PR head cannot be resolved — never proceed with
+    # an unpinned/unknown head even if candidate_head was supplied (#718).
+    if not live_head:
+        return {
+            "success": False,
+            "acquired": False,
+            "reasons": [
+                "live PR head could not be resolved from GET /pulls response "
+                "(exact-head scoping; fail closed)"
+            ],
+            "live_head": None,
+            "candidate_head": head_pin,
+        }
+    if live_head != head_pin:
+        return {
+            "success": False,
+            "acquired": False,
+            "reasons": [
+                f"candidate_head {head_pin[:12]}… does not match live PR head "
+                f"{live_head[:12]}… (exact-head scoping; fail closed)"
+            ],
+            "live_head": live_head,
+            "candidate_head": head_pin,
+        }
+
+    pr_merged_or_closed = bool(
+        pr_live.get("merged") or pr_live.get("merged_at")
+    ) or (str(pr_live.get("state") or "").strip().lower() == "closed")
+
+    assessment = reviewer_pr_lease.assess_acquire_lease(
+        comments,
+        pr_number=pr_number,
+        reviewer_identity=identity,
+        profile=profile.get("profile_name") or "unknown",
+        session_id=sid,
+        repo=repo_label,
+        issue_number=issue_number,
+        worktree=worktree,
+        candidate_head=head_pin,
+        target_branch=target_branch,
+        target_branch_sha=target_branch_sha,
+        pr_merged_or_closed=pr_merged_or_closed,
+    )
+    if not assessment.get("acquire_allowed"):
+        return {
+            "success": False,
+            "acquired": False,
+            "reasons": assessment.get("reasons") or [],
+            "existing_lease": assessment.get("existing_lease"),
+            "post_merge_moot": assessment.get("post_merge_moot", False),
+        }
+
+    body = assessment["lease_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": "acquire_merger_pr_lease"},
+    ):
+        posted = api_request("POST", comment_url, auth, {"body": body})
+
+    if not isinstance(posted, dict) or not posted.get("id"):
+        return {
+            "success": False,
+            "acquired": False,
+            "reasons": [
+                "lease comment POST failed or returned no comment id "
+                "(no partial session lease recorded)"
+            ],
+        }
+
+    session_lease = reviewer_pr_lease.record_session_lease({
+        "pr_number": pr_number,
+        "issue_number": issue_number,
+        "session_id": sid,
+        "reviewer_identity": identity,
+        "profile": profile.get("profile_name"),
+        "worktree": worktree,
+        "phase": "claimed",
+        "candidate_head": head_pin,
+        "target_branch": target_branch,
+        "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_MERGER,
+        comment_id=posted.get("id"),
+    ))
+    return {
+        "success": True,
+        "acquired": True,
+        "pr_number": pr_number,
+        "session_id": sid,
+        "comment_id": posted.get("id"),
+        "session_lease": session_lease,
+        "candidate_head": head_pin,
+        "repo": repo_label,
+        "reasons": [],
+    }
+
+
 @mcp.tool()
 def gitea_adopt_merger_pr_lease(
     pr_number: int,
@@ -10798,7 +11018,11 @@ def gitea_adopt_merger_pr_lease(
 
     # task=adopt_merger_pr_lease so verify_preflight_purity runs shared #604 anti-stomp.
     _verify_role_mutation_workspace(
-        remote, worktree=worktree, task="adopt_merger_pr_lease"
+        remote,
+        worktree=worktree,
+        task="adopt_merger_pr_lease",
+        org=org,
+        repo=repo,
     )
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
@@ -11520,7 +11744,11 @@ def gitea_release_reviewer_pr_lease(
         dict reporting release status.
     """
     _verify_role_mutation_workspace(
-        remote, worktree=worktree, task="review_pr"
+        remote,
+        worktree=worktree,
+        task="review_pr",
+        org=org,
+        repo=repo,
     )
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
@@ -14046,6 +14274,154 @@ def _branch_protection_requires_current_base(
     return None
 
 
+def _prove_author_ownership_for_pr(
+    *,
+    pr_number: int,
+    pr_title: str | None,
+    pr_body: str | None,
+    source_branch: str | None,
+    remote: str,
+    host: str | None,
+    org: str,
+    repo: str,
+    worktree_path: str | None = None,
+) -> dict:
+    """Prove author ownership for PR mutations without requiring issue==pr.
+
+    Issue #727 / PR #728: tracking issues and PR numbers often differ. Ownership
+    is proven via linked issues (title/body/branch), a live session lock for one
+    of those issues (or the PR number itself), a durable issue lock for a linked
+    issue, and optional branch/worktree binding on the lock. Unrelated issue
+    locks fail closed.
+    """
+    reasons: list[str] = []
+    text = f"{pr_title or ''}\n{pr_body or ''}"
+    linked = list(
+        extract_linked_issue_numbers(text, branch_name=source_branch)
+    )
+    # Always allow legacy same-number ownership when the lock is for the PR index.
+    candidates = sorted(set(linked + [int(pr_number)]))
+    remote_key = remote if remote in REMOTES else (host or "unknown")
+    matched_issue: int | None = None
+    matched_via: str | None = None
+    lock_record: dict | None = None
+
+    # 1) Session lock: active issue work lease for this session.
+    try:
+        session = issue_lock_store.read_session_issue_lock()
+    except Exception:
+        session = None
+    if session and issue_lock_store.is_lease_live(session):
+        try:
+            sess_issue = int(session.get("issue_number") or 0)
+        except (TypeError, ValueError):
+            sess_issue = 0
+        if sess_issue in candidates:
+            # Optional branch binding: when both sides declare a branch, they must match.
+            lock_branch = str(session.get("branch_name") or "").strip()
+            src = (source_branch or "").strip()
+            if lock_branch and src and lock_branch != src:
+                reasons.append(
+                    f"session lock for issue #{sess_issue} is bound to branch "
+                    f"'{lock_branch}', not PR source branch '{src}' (fail closed)"
+                )
+            else:
+                wt = (worktree_path or "").strip()
+                lock_wt = str(session.get("worktree_path") or "").strip()
+                if wt and lock_wt:
+                    try:
+                        same_wt = os.path.realpath(wt) == os.path.realpath(lock_wt)
+                    except Exception:
+                        same_wt = wt == lock_wt
+                    if not same_wt:
+                        reasons.append(
+                            f"session lock for issue #{sess_issue} is bound to a "
+                            "different worktree (fail closed)"
+                        )
+                    else:
+                        matched_issue = sess_issue
+                        matched_via = "session_lock"
+                        lock_record = dict(session)
+                else:
+                    matched_issue = sess_issue
+                    matched_via = "session_lock"
+                    lock_record = dict(session)
+        elif sess_issue:
+            reasons.append(
+                f"session lock issue #{sess_issue} is not linked to PR #{pr_number} "
+                f"(linked={linked or 'none'}; fail closed)"
+            )
+
+    # 2) Durable per-issue locks for each ownership candidate.
+    if matched_issue is None:
+        for issue_n in candidates:
+            try:
+                durable = issue_lock_store.load_issue_lock(
+                    remote=remote_key,
+                    org=org,
+                    repo=repo,
+                    issue_number=issue_n,
+                )
+            except Exception:
+                durable = None
+            if not durable or not issue_lock_store.is_lease_live(durable):
+                continue
+            lock_branch = str(durable.get("branch_name") or "").strip()
+            src = (source_branch or "").strip()
+            if lock_branch and src and lock_branch != src:
+                reasons.append(
+                    f"durable lock for issue #{issue_n} is bound to branch "
+                    f"'{lock_branch}', not PR source '{src}'"
+                )
+                continue
+            matched_issue = issue_n
+            matched_via = "durable_lock"
+            lock_record = dict(durable)
+            break
+
+    # 3) Branch-owned live lock (any issue) matching the PR source branch.
+    if matched_issue is None and (source_branch or "").strip():
+        try:
+            by_branch = issue_lock_store.find_live_lock_for_branch(
+                str(source_branch).strip()
+            )
+        except Exception:
+            by_branch = None
+        if by_branch:
+            try:
+                branch_issue = int(by_branch.get("issue_number") or 0)
+            except (TypeError, ValueError):
+                branch_issue = 0
+            if branch_issue in candidates:
+                matched_issue = branch_issue
+                matched_via = "branch_lock"
+                lock_record = dict(by_branch)
+            elif branch_issue:
+                reasons.append(
+                    f"live branch lock for '{source_branch}' belongs to unrelated "
+                    f"issue #{branch_issue} (not linked to PR #{pr_number}; fail closed)"
+                )
+
+    proven = matched_issue is not None and lock_record is not None
+    if not proven and not reasons:
+        reasons.append(
+            "no live author issue lock proven for PR "
+            f"#{pr_number} via linked issues {linked or '[]'}, session lock, "
+            "durable lock, or branch lock (fail closed; issue_number need not "
+            "equal pr_number when the tracking issue is linked)"
+        )
+    return {
+        "proven": proven,
+        "has_author_lock": proven,
+        "linked_issues": linked,
+        "ownership_candidates": candidates,
+        "matched_issue": matched_issue,
+        "matched_via": matched_via,
+        "lock_record": lock_record,
+        "reasons": reasons,
+    }
+
+
 @mcp.tool()
 def gitea_assess_pr_sync_status(
     pr_number: int,
@@ -14168,18 +14544,22 @@ def gitea_assess_pr_sync_status(
         approval_at_current_head = None
 
     # Locks / leases (best-effort; absence is reported as None/False).
+    # Ownership uses linked tracking issue / branch / session lock — not only
+    # issue_number == pr_number (#727 / PR #728).
     active_author_lock = None
     try:
-        lock = issue_lock_store.load_issue_lock(
-            remote=remote if remote in REMOTES else (h or "unknown"),
+        ownership = _prove_author_ownership_for_pr(
+            pr_number=pr_number,
+            pr_title=pr.get("title"),
+            pr_body=pr.get("body"),
+            source_branch=source_branch,
+            remote=remote,
+            host=host,
             org=o,
             repo=r,
-            issue_number=pr_number,
+            worktree_path=None,
         )
-        if lock:
-            active_author_lock = issue_lock_store.is_lease_live(lock)
-        else:
-            active_author_lock = False
+        active_author_lock = bool(ownership.get("has_author_lock"))
     except Exception:
         active_author_lock = None
 
@@ -14355,22 +14735,6 @@ def gitea_update_pr_branch_by_merge(
         or ""
     ).strip()
 
-    has_lock = False
-    try:
-        lock = issue_lock_store.read_session_issue_lock()
-        if lock and int(lock.get("issue_number") or 0) == int(pr_number):
-            has_lock = issue_lock_store.is_lease_live(lock)
-        if not has_lock:
-            lock2 = issue_lock_store.load_issue_lock(
-                remote=remote if remote in REMOTES else (h or "unknown"),
-                org=o,
-                repo=r,
-                issue_number=pr_number,
-            )
-            has_lock = bool(lock2 and issue_lock_store.is_lease_live(lock2))
-    except Exception:
-        has_lock = False
-
     auth = _auth(h)
     if not auth:
         return {
@@ -14448,6 +14812,20 @@ def gitea_update_pr_branch_by_merge(
         except Exception:
             owns_branch = None  # unknown — do not hard-fail solely on probe
 
+    # Prove ownership via linked issue lock / session / branch — not issue==pr.
+    ownership = _prove_author_ownership_for_pr(
+        pr_number=pr_number,
+        pr_title=pr.get("title"),
+        pr_body=pr.get("body"),
+        source_branch=source_branch,
+        remote=remote,
+        host=host,
+        org=o,
+        repo=r,
+        worktree_path=wt or None,
+    )
+    has_lock = bool(ownership.get("has_author_lock"))
+
     preflight = pr_sync_status.assess_update_pr_branch_preflight(
         role_kind=role,
         expected_pr_head_sha=expected_pr_head_sha,
@@ -14465,6 +14843,17 @@ def gitea_update_pr_branch_by_merge(
         force_push=False,
         rebase=False,
     )
+    if not has_lock and ownership.get("reasons"):
+        # Surface ownership diagnostics beyond the generic preflight line.
+        preflight.setdefault("reasons", [])
+        for reason in ownership["reasons"]:
+            if reason not in preflight["reasons"]:
+                preflight["reasons"].append(reason)
+    preflight["ownership"] = {
+        "linked_issues": ownership.get("linked_issues"),
+        "matched_issue": ownership.get("matched_issue"),
+        "matched_via": ownership.get("matched_via"),
+    }
     preflight["host"] = h
     preflight["org"] = o
     preflight["repo"] = r
@@ -15242,11 +15631,56 @@ def gitea_resolve_task_capability(
     TASK_MAP = task_capability_map.TASK_CAPABILITY_MAP
 
     if task not in TASK_MAP:
-        raise ValueError(f"Unknown task/action: '{task}' (fail closed)")
+        # #723: structured fail-closed unknown_task (never raise into internal_error).
+        profile = get_profile()
+        h = host or (REMOTES.get(remote, {}).get("host") if remote in REMOTES else None)
+        username = _authenticated_username(h) if h else None
+        active_role_kind = _profile_role_kind(profile)
+        switching = gitea_config.is_runtime_switching_enabled()
+        result = {
+            "requested_task": task,
+            "required_operation_permission": "unknown",
+            "required_role_kind": "unknown",
+            "active_profile": profile.get("profile_name", "unknown"),
+            "active_identity": username,
+            "active_role_kind": active_role_kind,
+            "active_profile_allowed_operations": profile.get("allowed_operations") or [],
+            "active_profile_permission_allowed": False,
+            "allowed_in_current_session": False,
+            "available_in_session": False,
+            "configured": False,
+            "restart_required": False,
+            "stop_required": True,
+            "task_role_guidance": [
+                f"STOP: Unknown task/action: '{task}' (fail closed)"
+            ],
+            "matching_configured_profile": [],
+            "runtime_switching_supported": switching,
+            "different_mcp_namespace_required": False,
+            "exact_safe_next_action": (
+                f"None; task '{task}' is unrecognized by the capability map."
+            ),
+            "reason": f"Unknown task/action: '{task}' (fail closed)",
+            "reason_code": "unknown_task",
+            "mutation_performed": False,
+        }
+        role_session_router.sync_route_from_capability(result)
+        was_terminal = capability_stop_terminal.is_active()
+        terminal = capability_stop_terminal.sync_from_capability_result(result)
+        if terminal:
+            result["terminal_mode"] = True
+            result["terminal_report"] = (
+                capability_stop_terminal.build_terminal_report(result)
+            )
+        elif was_terminal:
+            result["cleared_stale_denial"] = True
+        return result
 
     required_permission = task_capability_map.required_permission(task)
     required_role = task_capability_map.required_role(task)
     role_exclusive_tasks = {
+        "acquire_reviewer_pr_lease",
+        "gitea_acquire_reviewer_pr_lease",
         "review_pr",
         "approve_pr",
         "request_changes_pr",
@@ -15254,6 +15688,10 @@ def gitea_resolve_task_capability(
         "pr_queue_cleanup",
         "pr-queue-cleanup",
         "merge_pr",
+        "acquire_merger_pr_lease",
+        "gitea_acquire_merger_pr_lease",
+        "adopt_merger_pr_lease",
+        "gitea_adopt_merger_pr_lease",
         "create_branch",
         "push_branch",
         "create_pr",
diff --git a/merger_lease_adoption.py b/merger_lease_adoption.py
index 821779a..46ae926 100644
--- a/merger_lease_adoption.py
+++ b/merger_lease_adoption.py
@@ -18,11 +18,13 @@ DEFAULT_ADOPTION_REASON = "merger-handoff-approved-head"
 
 SOURCE_ADOPT = "gitea_adopt_merger_pr_lease"
 SOURCE_ACQUIRE = "gitea_acquire_reviewer_pr_lease"
+SOURCE_ACQUIRE_MERGER = "gitea_acquire_merger_pr_lease"
 SOURCE_HEARTBEAT = "gitea_heartbeat_reviewer_pr_lease"
 
 SANCTIONED_PROVENANCE_SOURCES = frozenset({
     SOURCE_ADOPT,
     SOURCE_ACQUIRE,
+    SOURCE_ACQUIRE_MERGER,
     SOURCE_HEARTBEAT,
 })
 
@@ -209,8 +211,10 @@ _SANCTIONED_LEASE_EVIDENCE_RE = re.compile(
     r"(?is)\b("
     r"gitea_adopt_merger_pr_lease|"
     r"gitea_acquire_reviewer_pr_lease|"
+    r"gitea_acquire_merger_pr_lease|"
     r"lease_proof_source\s*[:=]\s*gitea_adopt_merger_pr_lease|"
     r"lease_proof_source\s*[:=]\s*gitea_acquire_reviewer_pr_lease|"
+    r"lease_proof_source\s*[:=]\s*gitea_acquire_merger_pr_lease|"
     r"lease_proof_kind\s*[:=]\s*sanctioned_adoption|"
     r"lease_proof_kind\s*[:=]\s*sanctioned_acquire|"
     r"adoption_comment_id\s*[:=]|"
diff --git a/task_capability_map.py b/task_capability_map.py
index 516fd75..889ef3c 100644
--- a/task_capability_map.py
+++ b/task_capability_map.py
@@ -125,6 +125,18 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
         "permission": "gitea.pr.comment",
         "role": "merger",
     },
+    "gitea_adopt_merger_pr_lease": {
+        "permission": "gitea.pr.comment",
+        "role": "merger",
+    },
+    "acquire_merger_pr_lease": {
+        "permission": "gitea.pr.comment",
+        "role": "merger",
+    },
+    "gitea_acquire_merger_pr_lease": {
+        "permission": "gitea.pr.comment",
+        "role": "merger",
+    },
     # #691: guarded non-owner cleanup of obsolete comment-backed reviewer leases.
     # Apply path posts lease release + audit comments (gitea.pr.comment).
     "cleanup_obsolete_reviewer_comment_lease": {
diff --git a/tests/test_merger_lease_acquire.py b/tests/test_merger_lease_acquire.py
new file mode 100644
index 0000000..77bc118
--- /dev/null
+++ b/tests/test_merger_lease_acquire.py
@@ -0,0 +1,466 @@
+"""Merger lease acquisition + capability resolution tests (#718, #723)."""
+
+from __future__ import annotations
+
+import os
+import sys
+import unittest
+from datetime import datetime, timezone
+from unittest.mock import MagicMock, patch
+
+sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
+
+import gitea_mcp_server as mcp_server
+import reviewer_pr_lease as leases
+import task_capability_map as tcm
+
+
+HEAD = "a" * 40
+LIVE_HEAD = HEAD
+
+
+class TestMergerLeaseAcquireTool(unittest.TestCase):
+    def setUp(self):
+        mcp_server._preflight_whoami_called = True
+        mcp_server._preflight_capability_called = True
+        mcp_server._preflight_resolved_role = "merger"
+        mcp_server._preflight_resolved_task = "acquire_merger_pr_lease"
+        leases.clear_session_lease()
+
+    def tearDown(self):
+        leases.clear_session_lease()
+
+    def _merger_profile(self, **extra):
+        p = {
+            "username": "merger-user",
+            "profile_name": "prgs-merger",
+            "role": "merger",
+            "allowed_operations": [
+                "gitea.read",
+                "gitea.pr.comment",
+                "gitea.pr.merge",
+            ],
+            "forbidden_operations": [
+                "gitea.pr.approve",
+                "gitea.pr.review",
+                "gitea.pr.request_changes",
+            ],
+        }
+        p.update(extra)
+        return p
+
+    @patch("gitea_mcp_server.api_request")
+    @patch("gitea_mcp_server._auth", return_value="token pass")
+    @patch("gitea_mcp_server._resolve", return_value=("gitea.prgs.cc", "Scaled-Tech-Consulting", "Gitea-Tools"))
+    @patch("gitea_mcp_server.get_profile")
+    @patch("gitea_mcp_server._fetch_pr_comments", return_value=[])
+    @patch("gitea_mcp_server._verify_role_mutation_workspace")
+    def test_acquire_merger_lease_success(
+        self, _verify, _comments, mock_profile, _resolve, _auth, mock_api
+    ):
+        mock_profile.return_value = self._merger_profile()
+        mock_api.side_effect = [
+            {"state": "open", "head": {"sha": LIVE_HEAD}},
+            {"id": 456},
+        ]
+        with patch.dict(os.environ, {"GITEA_MCP_PROFILE_NAME": "prgs-merger"}):
+            res = mcp_server.gitea_acquire_merger_pr_lease(
+                pr_number=718,
+                worktree="branches/issue-718",
+                session_id="merger-session",
+                candidate_head=HEAD,
+                remote="prgs",
+                org="Scaled-Tech-Consulting",
+                repo="Gitea-Tools",
+            )
+        self.assertTrue(res["success"], res)
+        self.assertTrue(res["acquired"])
+        self.assertEqual(res["comment_id"], 456)
+        self.assertEqual(res["session_id"], "merger-session")
+        self.assertEqual(res.get("repo"), "Scaled-Tech-Consulting/Gitea-Tools")
+        body = mock_api.call_args_list[-1][0][3]["body"]
+        self.assertIn("reviewer_identity: merger-user", body)
+        self.assertIn("profile: prgs-merger", body)
+        session_lease = leases._SESSION_LEASE
+        self.assertIsNotNone(session_lease)
+        provenance = session_lease.get("lease_provenance") or {}
+        self.assertEqual(provenance.get("source"), "gitea_acquire_merger_pr_lease")
+
+    @patch("gitea_mcp_server.api_request")
+    @patch("gitea_mcp_server._auth", return_value="token pass")
+    @patch("gitea_mcp_server._resolve", return_value=("gitea.prgs.cc", "Scaled-Tech-Consulting", "Gitea-Tools"))
+    @patch("gitea_mcp_server.get_profile")
+    @patch("gitea_mcp_server._fetch_pr_comments")
+    @patch("gitea_mcp_server._verify_role_mutation_workspace")
+    def test_acquire_merger_lease_fails_if_active_exists(
+        self, _verify, _comments, mock_profile, _resolve, _auth, mock_api
+    ):
+        mock_profile.return_value = self._merger_profile()
+        active_body = leases.format_lease_body(
+            repo="Scaled-Tech-Consulting/Gitea-Tools",
+            pr_number=718,
+            issue_number=None,
+            reviewer_identity="other-user",
+            profile="prgs-reviewer",
+            session_id="other-session",
+            worktree="branches/review-1",
+            phase="claimed",
+            candidate_head=HEAD,
+            target_branch="master",
+            target_branch_sha="b" * 40,
+            last_activity=datetime.now(timezone.utc),
+        )
+        _comments.return_value = [
+            {"id": 1, "body": active_body, "user": {"login": "other-user"}}
+        ]
+        mock_api.return_value = {"state": "open", "head": {"sha": LIVE_HEAD}}
+        with patch.dict(os.environ, {"GITEA_MCP_PROFILE_NAME": "prgs-merger"}):
+            res = mcp_server.gitea_acquire_merger_pr_lease(
+                pr_number=718,
+                worktree="branches/issue-718",
+                session_id="merger-session",
+                candidate_head=HEAD,
+            )
+        self.assertFalse(res["success"])
+        self.assertFalse(res["acquired"])
+        self.assertIn("already has active reviewer lease", " ".join(res["reasons"]))
+        self.assertEqual(
+            [c for c in mock_api.call_args_list if c[0][0] == "POST"], []
+        )
+        self.assertIsNone(leases._SESSION_LEASE)
+
+    @patch("gitea_mcp_server.get_profile")
+    @patch("gitea_mcp_server._verify_role_mutation_workspace")
+    def test_acquire_merger_lease_fails_workspace_binding(self, _verify, mock_profile):
+        mock_profile.return_value = self._merger_profile()
+        _verify.side_effect = RuntimeError("Branches-only mutation guard")
+        with patch.dict(os.environ, {"GITEA_MCP_PROFILE_NAME": "prgs-merger"}):
+            res = mcp_server.gitea_acquire_merger_pr_lease(
+                pr_number=718,
+                worktree="branches/issue-718",
+                session_id="merger-session",
+                candidate_head=HEAD,
+            )
+        self.assertFalse(res["success"])
+        self.assertIn("Branches-only mutation guard", res["reasons"][0])
+        self.assertIsNone(leases._SESSION_LEASE)
+
+    @patch("gitea_mcp_server.api_request")
+    @patch("gitea_mcp_server._auth", return_value="token pass")
+    @patch("gitea_mcp_server._resolve", return_value=("gitea.prgs.cc", "Scaled-Tech-Consulting", "Gitea-Tools"))
+    @patch("gitea_mcp_server.get_profile")
+    @patch("gitea_mcp_server._fetch_pr_comments", return_value=[])
+    @patch("gitea_mcp_server._verify_role_mutation_workspace")
+    def test_acquire_merger_forwards_explicit_org_repo_to_workspace_verify(
+        self, mock_verify, _comments, mock_profile, _resolve, _auth, mock_api
+    ):
+        """Explicit org/repo must reach anti-stomp (prgs bare default is Timesheet)."""
+        mock_profile.return_value = self._merger_profile()
+        mock_api.side_effect = [
+            {"state": "open", "head": {"sha": LIVE_HEAD}},
+            {"id": 789},
+        ]
+        with patch.dict(os.environ, {"GITEA_MCP_PROFILE_NAME": "prgs-merger"}):
+            res = mcp_server.gitea_acquire_merger_pr_lease(
+                pr_number=718,
+                worktree="branches/issue-718",
+                session_id="merger-session",
+                candidate_head=HEAD,
+                remote="prgs",
+                org="Scaled-Tech-Consulting",
+                repo="Gitea-Tools",
+            )
+        self.assertTrue(res["success"], res)
+        kwargs = mock_verify.call_args.kwargs
+        self.assertEqual(kwargs.get("org"), "Scaled-Tech-Consulting")
+        self.assertEqual(kwargs.get("repo"), "Gitea-Tools")
+        self.assertEqual(kwargs.get("task"), "acquire_merger_pr_lease")
+
+    @patch("gitea_mcp_server.get_profile")
+    def test_acquire_merger_requires_candidate_head(self, mock_profile):
+        mock_profile.return_value = self._merger_profile()
+        res = mcp_server.gitea_acquire_merger_pr_lease(
+            pr_number=718,
+            worktree="branches/issue-718",
+            candidate_head=None,
+        )
+        self.assertFalse(res["success"])
+        self.assertTrue(any("candidate_head is required" in r for r in res["reasons"]))
+
+    @patch("gitea_mcp_server.api_request")
+    @patch("gitea_mcp_server._auth", return_value="token pass")
+    @patch("gitea_mcp_server._resolve", return_value=("gitea.prgs.cc", "Scaled-Tech-Consulting", "Gitea-Tools"))
+    @patch("gitea_mcp_server.get_profile")
+    @patch("gitea_mcp_server._fetch_pr_comments", return_value=[])
+    @patch("gitea_mcp_server._verify_role_mutation_workspace")
+    def test_acquire_merger_head_mismatch(
+        self, _verify, _comments, mock_profile, _resolve, _auth, mock_api
+    ):
+        mock_profile.return_value = self._merger_profile()
+        mock_api.return_value = {"state": "open", "head": {"sha": "b" * 40}}
+        res = mcp_server.gitea_acquire_merger_pr_lease(
+            pr_number=718,
+            worktree="branches/issue-718",
+            candidate_head=HEAD,
+        )
+        self.assertFalse(res["success"])
+        self.assertTrue(any("does not match live PR head" in r for r in res["reasons"]))
+        self.assertIsNone(leases._SESSION_LEASE)
+
+    @patch("gitea_mcp_server.api_request")
+    @patch("gitea_mcp_server._auth", return_value="token pass")
+    @patch("gitea_mcp_server._resolve", return_value=("gitea.prgs.cc", "Scaled-Tech-Consulting", "Gitea-Tools"))
+    @patch("gitea_mcp_server.get_profile")
+    @patch("gitea_mcp_server._fetch_pr_comments", return_value=[])
+    @patch("gitea_mcp_server._verify_role_mutation_workspace")
+    def test_acquire_merger_fails_closed_when_live_head_unresolvable(
+        self, _verify, _comments, mock_profile, _resolve, _auth, mock_api
+    ):
+        """Empty/missing live head must not silently proceed past exact-head gate."""
+        mock_profile.return_value = self._merger_profile()
+        # PR payload present but head.sha missing / empty / non-dict.
+        mock_api.return_value = {"state": "open", "head": {}}
+        res = mcp_server.gitea_acquire_merger_pr_lease(
+            pr_number=718,
+            worktree="branches/issue-718",
+            session_id="merger-session",
+            candidate_head=HEAD,
+        )
+        self.assertFalse(res["success"], res)
+        self.assertFalse(res.get("acquired"), res)
+        self.assertTrue(
+            any("live PR head could not be resolved" in r for r in res["reasons"]),
+            res,
+        )
+        self.assertIsNone(res.get("live_head"))
+        self.assertEqual(res.get("candidate_head"), HEAD)
+        self.assertEqual(
+            [c for c in mock_api.call_args_list if c[0][0] == "POST"], []
+        )
+        self.assertIsNone(leases._SESSION_LEASE)
+
+        # Completely empty GET /pulls body also fails closed.
+        mock_api.return_value = {}
+        res2 = mcp_server.gitea_acquire_merger_pr_lease(
+            pr_number=718,
+            worktree="branches/issue-718",
+            candidate_head=HEAD,
+        )
+        self.assertFalse(res2["success"], res2)
+        self.assertTrue(
+            any("live PR head could not be resolved" in r for r in res2["reasons"]),
+            res2,
+        )
+        self.assertIsNone(leases._SESSION_LEASE)
+
+    @patch("gitea_mcp_server._profile_operation_gate")
+    def test_author_denied_merge_permission(self, mock_gate):
+        """Author profile lacks gitea.pr.merge → fail closed before mutation."""
+        def gate(op):
+            if op == "gitea.pr.merge":
+                return [f"profile lacks {op}"]
+            return None
+        mock_gate.side_effect = gate
+        res = mcp_server.gitea_acquire_merger_pr_lease(
+            pr_number=718,
+            worktree="branches/issue-718",
+            candidate_head=HEAD,
+        )
+        self.assertFalse(res["success"])
+        self.assertTrue(any("gitea.pr.merge" in r for r in res["reasons"]))
+
+    @patch("gitea_mcp_server._profile_operation_gate")
+    def test_reviewer_denied_merge_permission(self, mock_gate):
+        def gate(op):
+            if op == "gitea.pr.merge":
+                return [f"profile lacks {op}"]
+            return None
+        mock_gate.side_effect = gate
+        res = mcp_server.gitea_acquire_merger_pr_lease(
+            pr_number=718,
+            worktree="branches/issue-718",
+            candidate_head=HEAD,
+        )
+        self.assertFalse(res["success"])
+        self.assertTrue(any("gitea.pr.merge" in r for r in res["reasons"]))
+
+
+class TestCapabilityMapLeaseTasks(unittest.TestCase):
+    def test_merger_acquire_map_entries(self):
+        for task in ("acquire_merger_pr_lease", "gitea_acquire_merger_pr_lease"):
+            self.assertEqual(tcm.required_permission(task), "gitea.pr.comment")
+            self.assertEqual(tcm.required_role(task), "merger")
+
+    def test_reviewer_acquire_map_entries(self):
+        for task in ("acquire_reviewer_pr_lease", "gitea_acquire_reviewer_pr_lease"):
+            self.assertEqual(tcm.required_permission(task), "gitea.pr.comment")
+            self.assertEqual(tcm.required_role(task), "reviewer")
+
+    def test_adopt_merger_aliases(self):
+        for task in ("adopt_merger_pr_lease", "gitea_adopt_merger_pr_lease"):
+            self.assertEqual(tcm.required_role(task), "merger")
+
+
+class TestResolveTaskCapabilityLeaseAndUnknown(unittest.TestCase):
+    def setUp(self):
+        mcp_server._process_boot_head_sha = None
+        if hasattr(mcp_server, "capability_stop_terminal"):
+            mcp_server.capability_stop_terminal.clear()
+
+    @patch.object(mcp_server, "record_mutation_authority", return_value=None)
+    @patch.object(mcp_server, "init_review_decision_lock", return_value=None)
+    @patch.object(mcp_server, "record_preflight_check", return_value=None)
+    @patch.object(mcp_server, "_ensure_matching_profile", return_value=None)
+    @patch.object(mcp_server, "_authenticated_username", return_value="sysadmin")
+    @patch.object(mcp_server, "get_profile")
+    @patch.object(mcp_server.gitea_config, "load_config")
+    def test_resolve_acquire_reviewer_authorized(
+        self, mock_cfg, mock_profile, *_mocks
+    ):
+        mock_profile.return_value = {
+            "profile_name": "prgs-reviewer",
+            "role": "reviewer",
+            "allowed_operations": ["gitea.pr.comment", "gitea.read", "gitea.pr.review"],
+            "forbidden_operations": [],
+        }
+        mock_cfg.return_value = {
+            "profiles": {
+                "prgs-reviewer": {
+                    "role": "reviewer",
+                    "allowed_operations": [
+                        "gitea.pr.comment",
+                        "gitea.read",
+                        "gitea.pr.review",
+                    ],
+                    "forbidden_operations": [],
+                }
+            }
+        }
+        with patch.dict(os.environ, {"GITEA_MCP_PROFILE": "prgs-reviewer"}, clear=False):
+            for task in ("acquire_reviewer_pr_lease", "gitea_acquire_reviewer_pr_lease"):
+                res = mcp_server.gitea_resolve_task_capability(task=task, remote="prgs")
+                self.assertEqual(res["required_role_kind"], "reviewer", res)
+                self.assertEqual(
+                    res["required_operation_permission"], "gitea.pr.comment", res
+                )
+                self.assertTrue(res.get("allowed_in_current_session"), res)
+
+    @patch.object(mcp_server, "record_mutation_authority", return_value=None)
+    @patch.object(mcp_server, "init_review_decision_lock", return_value=None)
+    @patch.object(mcp_server, "record_preflight_check", return_value=None)
+    @patch.object(mcp_server, "_ensure_matching_profile", return_value=None)
+    @patch.object(mcp_server, "_authenticated_username", return_value="jcwalker3")
+    @patch.object(mcp_server, "get_profile")
+    @patch.object(mcp_server.gitea_config, "load_config")
+    def test_resolve_acquire_reviewer_author_denied(
+        self, mock_cfg, mock_profile, *_mocks
+    ):
+        mock_profile.return_value = {
+            "profile_name": "prgs-author",
+            "role": "author",
+            "allowed_operations": ["gitea.pr.comment", "gitea.branch.push", "gitea.read"],
+            "forbidden_operations": [],
+        }
+        mock_cfg.return_value = {
+            "profiles": {
+                "prgs-author": {
+                    "role": "author",
+                    "allowed_operations": [
+                        "gitea.pr.comment",
+                        "gitea.branch.push",
+                        "gitea.read",
+                    ],
+                    "forbidden_operations": [],
+                },
+                "prgs-reviewer": {
+                    "role": "reviewer",
+                    "allowed_operations": ["gitea.pr.comment", "gitea.pr.review"],
+                    "forbidden_operations": [],
+                },
+            }
+        }
+        with patch.dict(os.environ, {"GITEA_MCP_PROFILE": "prgs-author"}, clear=False):
+            res = mcp_server.gitea_resolve_task_capability(
+                task="acquire_reviewer_pr_lease", remote="prgs"
+            )
+        self.assertFalse(res.get("allowed_in_current_session"), res)
+        self.assertEqual(res["required_role_kind"], "reviewer")
+        guidance = " ".join(res.get("task_role_guidance") or [])
+        self.assertTrue(
+            "cannot perform reviewer" in guidance.lower()
+            or "reviewer" in guidance.lower()
+            or res.get("stop_required"),
+            res,
+        )
+
+    @patch.object(mcp_server, "record_mutation_authority", return_value=None)
+    @patch.object(mcp_server, "init_review_decision_lock", return_value=None)
+    @patch.object(mcp_server, "record_preflight_check", return_value=None)
+    @patch.object(mcp_server, "_ensure_matching_profile", return_value=None)
+    @patch.object(mcp_server, "_authenticated_username", return_value="sysadmin")
+    @patch.object(mcp_server, "get_profile")
+    def test_resolve_unknown_task_structured_no_raise(self, mock_profile, *_mocks):
+        mock_profile.return_value = {
+            "profile_name": "prgs-reviewer",
+            "role": "reviewer",
+            "allowed_operations": ["gitea.pr.comment"],
+            "forbidden_operations": [],
+        }
+        # Must not raise ValueError into the tool boundary.
+        res = mcp_server.gitea_resolve_task_capability(
+            task="some_made_up_task", remote="prgs"
+        )
+        self.assertIsInstance(res, dict)
+        self.assertEqual(res.get("reason_code"), "unknown_task", res)
+        self.assertFalse(res.get("allowed_in_current_session"))
+        self.assertTrue(res.get("stop_required"))
+        self.assertIs(res.get("mutation_performed"), False)
+        self.assertNotIn("token", str(res).lower())
+        self.assertNotIn("password", str(res).lower())
+        guidance = " ".join(res.get("task_role_guidance") or [])
+        self.assertIn("Unknown task/action", guidance)
+
+    @patch.object(mcp_server, "record_mutation_authority", return_value=None)
+    @patch.object(mcp_server, "init_review_decision_lock", return_value=None)
+    @patch.object(mcp_server, "record_preflight_check", return_value=None)
+    @patch.object(mcp_server, "_ensure_matching_profile", return_value=None)
+    @patch.object(mcp_server, "_authenticated_username", return_value="sysadmin")
+    @patch.object(mcp_server, "get_profile")
+    @patch.object(mcp_server.gitea_config, "load_config")
+    def test_resolve_merger_acquire_aliases(
+        self, mock_cfg, mock_profile, *_mocks
+    ):
+        mock_profile.return_value = {
+            "profile_name": "prgs-merger",
+            "role": "merger",
+            "allowed_operations": [
+                "gitea.read",
+                "gitea.pr.comment",
+                "gitea.pr.merge",
+            ],
+            "forbidden_operations": [],
+        }
+        mock_cfg.return_value = {
+            "profiles": {
+                "prgs-merger": {
+                    "role": "merger",
+                    "allowed_operations": [
+                        "gitea.read",
+                        "gitea.pr.comment",
+                        "gitea.pr.merge",
+                    ],
+                    "forbidden_operations": [],
+                }
+            }
+        }
+        with patch.dict(os.environ, {"GITEA_MCP_PROFILE": "prgs-merger"}, clear=False):
+            for task in ("acquire_merger_pr_lease", "gitea_acquire_merger_pr_lease"):
+                res = mcp_server.gitea_resolve_task_capability(task=task, remote="prgs")
+                self.assertEqual(res["required_role_kind"], "merger", res)
+                self.assertEqual(
+                    res["required_operation_permission"], "gitea.pr.comment", res
+                )
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/tests/test_pr_ownership_issue_pr_mismatch.py b/tests/test_pr_ownership_issue_pr_mismatch.py
new file mode 100644
index 0000000..2286dae
--- /dev/null
+++ b/tests/test_pr_ownership_issue_pr_mismatch.py
@@ -0,0 +1,196 @@
+"""#727 / PR #728: author ownership when tracking issue number != PR number.
+
+Regression for REQUEST_CHANGES: gitea_update_pr_branch_by_merge must not require
+issue_number == pr_number. Ownership is proven via linked issue + lock/branch
+context and fails closed for unrelated issues.
+"""
+
+from __future__ import annotations
+
+import os
+import tempfile
+import unittest
+from datetime import datetime, timedelta, timezone
+from unittest.mock import patch
+
+import issue_lock_store
+import gitea_mcp_server as mcp
+
+
+def _live_lock(
+    *,
+    issue_number: int,
+    branch_name: str = "feat/issue-727-pr-sync-status",
+    worktree_path: str = "/tmp/branches/issue-727-pr-sync-status",
+    remote: str = "prgs",
+    org: str = "Scaled-Tech-Consulting",
+    repo: str = "Gitea-Tools",
+) -> dict:
+    now = datetime.now(timezone.utc)
+    return {
+        "issue_number": issue_number,
+        "branch_name": branch_name,
+        "worktree_path": worktree_path,
+        "remote": remote,
+        "org": org,
+        "repo": repo,
+        "operation_type": issue_lock_store.AUTHOR_ISSUE_WORK_LEASE,
+        "acquired_at": now.isoformat(),
+        "expires_at": (now + timedelta(hours=2)).isoformat(),
+        "owner_pid": os.getpid(),
+        "status": "active",
+    }
+
+
+class TestAuthorOwnershipIssuePrMismatch(unittest.TestCase):
+    def setUp(self):
+        self._tmp = tempfile.TemporaryDirectory(prefix="gitea-issue-locks-")
+        self.lock_dir = self._tmp.name
+        os.environ["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir
+
+    def tearDown(self):
+        # Clear session pointer for this process.
+        try:
+            ptr = issue_lock_store.session_pointer_path(self.lock_dir)
+            if os.path.isfile(ptr):
+                os.unlink(ptr)
+        except Exception:
+            pass
+        os.environ.pop("GITEA_ISSUE_LOCK_DIR", None)
+        self._tmp.cleanup()
+
+    def test_issue_727_pr_728_session_lock_accepted(self):
+        """Tracking issue #727 lock is valid ownership for PR #728."""
+        lock = _live_lock(issue_number=727)
+        issue_lock_store.bind_session_lock(lock)
+        result = mcp._prove_author_ownership_for_pr(
+            pr_number=728,
+            pr_title="feat: pr sync",
+            pr_body="Closes #727\n\nNative PR sync lifecycle.",
+            source_branch="feat/issue-727-pr-sync-status",
+            remote="prgs",
+            host=None,
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+            worktree_path=lock["worktree_path"],
+        )
+        self.assertTrue(result["proven"], result)
+        self.assertTrue(result["has_author_lock"])
+        self.assertEqual(result["matched_issue"], 727)
+        self.assertEqual(result["matched_via"], "session_lock")
+        self.assertIn(727, result["linked_issues"])
+
+    def test_issue_727_pr_728_durable_lock_accepted(self):
+        lock = _live_lock(issue_number=727)
+        path = issue_lock_store.lock_file_path(
+            remote="prgs",
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+            issue_number=727,
+            lock_dir=self.lock_dir,
+        )
+        issue_lock_store.save_lock_file(path, lock)
+        result = mcp._prove_author_ownership_for_pr(
+            pr_number=728,
+            pr_title="feat: pr sync",
+            pr_body="Fixes #727",
+            source_branch="feat/issue-727-pr-sync-status",
+            remote="prgs",
+            host=None,
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+        )
+        self.assertTrue(result["proven"], result)
+        self.assertEqual(result["matched_issue"], 727)
+        self.assertEqual(result["matched_via"], "durable_lock")
+
+    def test_legacy_same_number_still_accepted(self):
+        lock = _live_lock(
+            issue_number=100,
+            branch_name="fix/issue-100",
+            worktree_path="/tmp/branches/issue-100",
+        )
+        issue_lock_store.bind_session_lock(lock)
+        result = mcp._prove_author_ownership_for_pr(
+            pr_number=100,
+            pr_title="fix something",
+            pr_body="Closes #100",
+            source_branch="fix/issue-100",
+            remote="prgs",
+            host=None,
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+            worktree_path=lock["worktree_path"],
+        )
+        self.assertTrue(result["proven"], result)
+        self.assertEqual(result["matched_issue"], 100)
+
+    def test_unrelated_issue_lock_rejected(self):
+        """Lock for issue #999 must not authorize PR #728 linked only to #727."""
+        lock = _live_lock(
+            issue_number=999,
+            branch_name="feat/issue-999",
+            worktree_path="/tmp/branches/issue-999",
+        )
+        issue_lock_store.bind_session_lock(lock)
+        path = issue_lock_store.lock_file_path(
+            remote="prgs",
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+            issue_number=999,
+            lock_dir=self.lock_dir,
+        )
+        issue_lock_store.save_lock_file(path, lock)
+        result = mcp._prove_author_ownership_for_pr(
+            pr_number=728,
+            pr_title="feat: pr sync",
+            pr_body="Closes #727",
+            source_branch="feat/issue-727-pr-sync-status",
+            remote="prgs",
+            host=None,
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+            worktree_path=lock["worktree_path"],
+        )
+        self.assertFalse(result["proven"], result)
+        self.assertFalse(result["has_author_lock"])
+        self.assertIsNone(result["matched_issue"])
+        self.assertTrue(result["reasons"])
+
+    def test_branch_mismatch_on_session_lock_fail_closed(self):
+        lock = _live_lock(
+            issue_number=727,
+            branch_name="feat/other-branch",
+        )
+        issue_lock_store.bind_session_lock(lock)
+        result = mcp._prove_author_ownership_for_pr(
+            pr_number=728,
+            pr_title="feat: pr sync",
+            pr_body="Closes #727",
+            source_branch="feat/issue-727-pr-sync-status",
+            remote="prgs",
+            host=None,
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+            worktree_path=lock["worktree_path"],
+        )
+        self.assertFalse(result["proven"], result)
+        self.assertTrue(any("branch" in r for r in result["reasons"]))
+
+    def test_no_lock_fail_closed(self):
+        result = mcp._prove_author_ownership_for_pr(
+            pr_number=728,
+            pr_title="feat: pr sync",
+            pr_body="Closes #727",
+            source_branch="feat/issue-727-pr-sync-status",
+            remote="prgs",
+            host=None,
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+        )
+        self.assertFalse(result["proven"], result)
+        self.assertTrue(any("no live author issue lock" in r for r in result["reasons"]))
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/tests/test_resolve_task_capability.py b/tests/test_resolve_task_capability.py
index 23f05bc..07e5769 100644
--- a/tests/test_resolve_task_capability.py
+++ b/tests/test_resolve_task_capability.py
@@ -231,9 +231,16 @@ class TestResolveTaskCapability(unittest.TestCase):
             self.assertIn("gitea.issue.comment", res["active_profile_allowed_operations"])
 
     def test_resolve_unknown_task_fails_closed(self):
+        # #723: unknown tasks return structured unknown_task (no ValueError escape).
         with patch.dict(os.environ, self._env("author-profile")):
-            with self.assertRaises(ValueError):
-                mcp_server.gitea_resolve_task_capability(task="invalid_task_xyz", remote="prgs")
+            res = mcp_server.gitea_resolve_task_capability(
+                task="invalid_task_xyz", remote="prgs"
+            )
+            self.assertIsInstance(res, dict)
+            self.assertEqual(res.get("reason_code"), "unknown_task", res)
+            self.assertFalse(res.get("allowed_in_current_session"))
+            self.assertTrue(res.get("stop_required"))
+            self.assertIs(res.get("mutation_performed"), False)
 
     # Additional regression tests per #145 for permission boundaries and structured guidance
     def test_issue_comment_does_not_imply_close(self):
@@ -439,8 +446,11 @@ class TestResolveTaskCapability(unittest.TestCase):
             res = mcp_server.gitea_resolve_task_capability(task="close_pr", remote="prgs")
             self.assertEqual(res["required_operation_permission"], "gitea.pr.close")
             for unknown in ("close_pull_request", "close", "pr_close"):
-                with self.assertRaises(ValueError):
-                    mcp_server.gitea_resolve_task_capability(task=unknown, remote="prgs")
+                unk = mcp_server.gitea_resolve_task_capability(
+                    task=unknown, remote="prgs"
+                )
+                self.assertEqual(unk.get("reason_code"), "unknown_task", unk)
+                self.assertFalse(unk.get("allowed_in_current_session"))
 
 if __name__ == "__main__":
     unittest.main()