fix(mcp): chain-scoped newest-wins reviewer lease reads in pr_work_lease (#742)

Second author remediation on PR #743. Fixes the full-ledger defect surfaced by
the first remediation.

Root cause: pr_work_lease.find_active_reviewer_lease iterated the reviewer
markers newest-first and used `continue` on a terminal marker. On a realistic
append-only ledger (claimed -> validating -> terminal) it therefore stepped
over the terminal marker and returned the older claim of the very chain that
marker had just ended. reviewer_pr_lease got strict newest-wins under #577;
pr_work_lease never did, so the two modules disagreed for released, blocked,
done, and abandoned alike, and merger owner finalization did not leave "no
active lease" for pr_work_lease consumers (conflict-fix acquire, PR sync
inventory).

Fix: a claim is skipped when a later marker terminates its own chain. Chain
identity is (repo, pr_number, candidate_head, session_id, reviewer_identity,
profile) via the new _reviewer_chain_key; _chain_terminated_after scans only
markers appended after the candidate. Consequences:

- a valid terminal marker ends its matching earlier claim (no resurrection);
- a foreign-session, wrong-repo, wrong-PR, wrong-head, wrong-identity, or
  wrong-profile terminal marker cannot cancel another session's active lease,
  which strict newest-wins alone would have allowed;
- a malformed terminal marker has no provable chain key, so it cancels nothing
  and cannot hide a valid active claim;
- expiry, freshness, phase sets, ownership and integrity checks, both parsers,
  and find_active_conflict_fix_lease are untouched;
- history stays append-only; no marker is deleted or rewritten.

Reviewer lease markers carry no token field, so token-fingerprint validation
remains where it already lives (session provenance, merger finalization) and is
not weakened here.

Tests: new TestFullLedgerNewestWins in tests/test_merger_lease_finalization.py
covers claimed->released/blocked/done/abandoned, claimed->validating->terminal,
foreign-session and mismatched repo/PR/head/identity/profile terminals, the
malformed terminal marker, a newer active chain surviving an older terminated
chain, newest-valid-chain selection across multiple histories, all-chains-
terminated, single-marker parity between the two modules across eight phases,
expired-claim/freshness non-regression, and the real ledger written by
gitea_release_merger_pr_lease reading as terminal in both modules with the
prior marker preserved. TestCrossModuleTerminalPhaseAgreement's scope-limited
placeholder test is replaced by a real both-modules full-ledger assertion.

Verified 10 of the new tests fail at the prior head 7ae5f3a and pass here.

Validation: focused TestFullLedgerNewestWins 14 passed / 21 subtests;
tests/test_merger_lease_finalization.py 59 passed / 42 subtests; targeted
pr_work_lease + reviewer/merger lease + adoption + provenance + acquire/release
+ anti-stomp + capability-map + decision-lock + report-validator suites 309
passed / 41 subtests; full suite 3326 passed, 6 skipped, 2 failed. Both
failures (test_issue_702_review_findings_f1_f6 F1 recovery, reconciler
supersession close) reproduce identically on clean baseline master a8d2087 and
are pre-existing #737 org/repo-forwarding drift. git diff --check clean.

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

gitea_mcp_server.py

@@ -11585,6 +11585,11 @@ def gitea_acquire_merger_pr_lease(
     }, lease_provenance=merger_lease_adoption.build_lease_provenance(
         source=merger_lease_adoption.SOURCE_ACQUIRE_MERGER,
         comment_id=posted.get("id"),
+        # #742: bind the lease to this native runtime so a later owner-session
+        # finalization can prove it is the same daemon that acquired it.
+        native_token_fingerprint=(
+            mcp_daemon_guard.native_runtime_status().get("token_fingerprint")
+        ),
     ))
     return {
         "success": True,
@@ -11774,6 +11779,191 @@ def gitea_adopt_merger_pr_lease(
     }
 
 
+@mcp.tool()
+def gitea_release_merger_pr_lease(
+    pr_number: int,
+    worktree: str,
+    candidate_head: str,
+    outcome: str = merger_lease_adoption.OUTCOME_RELEASED,
+    reason: str | None = None,
+    issue_number: int | None = None,
+    remote: str = "dadeschools",
+    host: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+) -> dict:
+    """Terminally release or abandon this merger session's own PR lease (#742).
+
+    Sanctioned owner-session finalization for a comment-backed merger lease when
+    the merge does **not** occur (non-retryable merge failure, aborted merge, or
+    a blocker that ends the attempt). Without it a failed merger lease can only
+    expire, blocking the queue for the remainder of its TTL.
+
+    Merger-only and owner-only. The caller must hold the exact in-session lease
+    it is finalizing; foreign-session release, reviewer profiles, and mismatched
+    repository / PR / candidate head / identity / profile / native runtime token
+    fingerprint all fail closed. Reviewer sessions use
+    ``gitea_release_reviewer_pr_lease`` instead — that tool is never a merger
+    path.
+
+    Finalization is append-only: it posts a terminal lease marker and never
+    edits or deletes prior ledger comments. It is idempotent — an already
+    terminal lease returns ``already_terminal`` and posts nothing. The PR, its
+    approval, the review decision lock, the branch, and the worktree are all
+    preserved; only the lease is ended.
+
+    Args:
+        pr_number: PR whose merger lease to finalize.
+        outcome: 'released' (clean end) or 'abandoned' (ended under a blocker).
+        reason: Short finalization reason recorded in the marker.
+        candidate_head: Pinned head SHA the lease was acquired against.
+
+    Returns:
+        dict reporting finalization status, reasons, and the marker comment id.
+    """
+    read_block = _profile_operation_gate("gitea.read")
+    if read_block:
+        return {
+            "success": False,
+            "finalized": 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,
+            "finalized": False,
+            "reasons": comment_block,
+            "permission_report": _permission_block_report("gitea.pr.comment"),
+        }
+    # Merger-role proof: reviewer profiles hold gitea.pr.comment but never
+    # gitea.pr.merge, so this gate keeps the tool merger-only (#742).
+    merge_block = _profile_operation_gate("gitea.pr.merge")
+    if merge_block:
+        return {
+            "success": False,
+            "finalized": False,
+            "reasons": merge_block,
+            "permission_report": _permission_block_report("gitea.pr.merge"),
+        }
+
+    try:
+        _verify_role_mutation_workspace(
+            remote,
+            worktree=worktree,
+            task="release_merger_pr_lease",
+            org=org,
+            repo=repo,
+        )
+    except RuntimeError as e:
+        return {
+            "success": False,
+            "finalized": False,
+            "reasons": [str(e)],
+        }
+
+    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(h) or profile.get("username") or ""
+    repo_label = f"{o}/{r}"
+
+    session = reviewer_pr_lease.get_session_lease()
+    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 = str(
+        (pr_live.get("head") or {}).get("sha") or pr_live.get("head_sha") or ""
+    ).strip()
+
+    assessment = merger_lease_adoption.assess_merger_lease_finalization(
+        comments,
+        pr_number=pr_number,
+        session=session,
+        actor_identity=identity,
+        actor_profile=profile_name,
+        actor_session_id=(session or {}).get("session_id"),
+        repo=repo_label,
+        worktree=worktree,
+        candidate_head=candidate_head,
+        live_head_sha=live_head or None,
+        outcome=outcome,
+        reason=reason,
+        runtime_token_fingerprint=(
+            mcp_daemon_guard.native_runtime_status().get("token_fingerprint")
+        ),
+        issue_number=issue_number,
+    )
+
+    if assessment.get("already_terminal"):
+        # Idempotent: the lease is already terminal on the ledger. Clear the
+        # in-session mirror so no stale proof survives, and post nothing.
+        reviewer_pr_lease.clear_session_lease()
+        return {
+            "success": True,
+            "finalized": False,
+            "already_terminal": True,
+            "pr_number": pr_number,
+            "outcome": assessment.get("outcome"),
+            "reasons": [],
+        }
+
+    if not assessment.get("finalize_allowed"):
+        return {
+            "success": False,
+            "finalized": False,
+            "already_terminal": False,
+            "pr_number": pr_number,
+            "reasons": assessment.get("reasons") or [],
+        }
+
+    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_merger_pr_lease"},
+    ):
+        posted = api_request(
+            "POST", comment_url, auth, {"body": assessment["finalization_body"]}
+        )
+
+    if not isinstance(posted, dict) or not posted.get("id"):
+        # The lease is still live on the ledger: keep the in-session mirror so
+        # the owner can retry finalization rather than losing its own proof.
+        return {
+            "success": False,
+            "finalized": False,
+            "reasons": [
+                "terminal lease marker POST failed or returned no comment id "
+                "(lease left intact; retry finalization)"
+            ],
+        }
+
+    reviewer_pr_lease.clear_session_lease()
+
+    return {
+        "success": True,
+        "finalized": True,
+        "already_terminal": False,
+        "pr_number": pr_number,
+        "outcome": assessment.get("outcome"),
+        "finalization_reason": assessment.get("finalization_reason"),
+        "finalization_comment_id": posted.get("id"),
+        "finalized_lease_comment_id": assessment.get("lease_comment_id"),
+        "candidate_head": assessment.get("candidate_head"),
+        "repo": repo_label,
+        "reasons": [],
+    }
+
+
 @mcp.tool()
 def gitea_heartbeat_reviewer_pr_lease(
     pr_number: int,
@@ -16482,6 +16672,8 @@ def gitea_resolve_task_capability(
         "gitea_acquire_merger_pr_lease",
         "adopt_merger_pr_lease",
         "gitea_adopt_merger_pr_lease",
+        "release_merger_pr_lease",
+        "gitea_release_merger_pr_lease",
         "create_branch",
         "push_branch",
         "create_pr",

merger_lease_adoption.py

@@ -20,6 +20,7 @@ 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"
+SOURCE_RELEASE_MERGER = "gitea_release_merger_pr_lease"
 
 SANCTIONED_PROVENANCE_SOURCES = frozenset({
     SOURCE_ADOPT,
@@ -30,6 +31,21 @@ SANCTIONED_PROVENANCE_SOURCES = frozenset({
 
 _MERGER_ADOPTABLE_FRESHNESS = frozenset({"active", "stale_warning"})
 
+# #742: owner-session terminal finalization of a merger-held lease. Append-only
+# marker; ledger history is never edited or deleted.
+MERGER_FINALIZATION_MARKER = "<!-- mcp-merger-lease-final:v1 -->"
+OUTCOME_RELEASED = "released"
+OUTCOME_ABANDONED = "abandoned"
+MERGER_FINALIZATION_OUTCOMES = frozenset({OUTCOME_RELEASED, OUTCOME_ABANDONED})
+DEFAULT_MERGER_FINALIZATION_REASON = "merge-not-performed"
+
+# Provenance sources whose in-session lease is merger-owned and therefore
+# finalizable by its owning merger session.
+MERGER_OWNED_PROVENANCE_SOURCES = frozenset({
+    SOURCE_ACQUIRE_MERGER,
+    SOURCE_ADOPT,
+})
+
 
 def format_adoption_body(
     *,
@@ -97,6 +113,7 @@ def build_lease_provenance(
     adopted_from_profile: str | None = None,
     adopted_from_reviewer_identity: str | None = None,
     adoption_reason: str | None = None,
+    native_token_fingerprint: str | None = None,
     recorded_at: datetime | None = None,
 ) -> dict[str, Any]:
     recorded_at = recorded_at or datetime.now(timezone.utc)
@@ -117,9 +134,76 @@ def build_lease_provenance(
         proof["adopted_from_reviewer_identity"] = adopted_from_reviewer_identity
     if adoption_reason:
         proof["adoption_reason"] = adoption_reason
+    if native_token_fingerprint:
+        proof["native_token_fingerprint"] = native_token_fingerprint
     return proof
 
 
+def assess_acquired_merger_lease_integrity(
+    session: dict[str, Any] | None,
+) -> list[str]:
+    """Ownership/integrity reasons blocking a ``SOURCE_ACQUIRE_MERGER`` lease (#742).
+
+    A merger lease minted by ``gitea_acquire_merger_pr_lease`` authorizes an
+    irreversible merge, so it is sanctioned only when the in-session record is
+    complete and self-consistent: comment marker, exact session identity, merger
+    profile/role, repository, PR number, and pinned candidate head. Any missing
+    or contradictory field fails closed — an incomplete record is not proof.
+    """
+    if not session:
+        return ["no in-session lease recorded"]
+    provenance = session.get("lease_provenance") or {}
+    if not isinstance(provenance, dict):
+        provenance = {}
+    reasons: list[str] = []
+
+    provenance_comment_id = provenance.get("comment_id")
+    session_comment_id = session.get("comment_id")
+    if not (provenance_comment_id or session_comment_id):
+        reasons.append(
+            "acquired merger lease has no comment marker id (comment-backed "
+            "proof required)"
+        )
+    elif (
+        provenance_comment_id is not None
+        and session_comment_id is not None
+        and provenance_comment_id != session_comment_id
+    ):
+        reasons.append(
+            "acquired merger lease provenance comment_id does not match the "
+            "session lease comment_id"
+        )
+
+    if not (session.get("session_id") or "").strip():
+        reasons.append("acquired merger lease has no session_id")
+    if not (session.get("reviewer_identity") or "").strip():
+        reasons.append("acquired merger lease has no holder identity")
+
+    profile = (session.get("profile") or "").strip()
+    if not profile:
+        reasons.append("acquired merger lease has no profile")
+    elif "merger" not in profile.lower():
+        reasons.append(
+            f"acquired merger lease profile '{profile}' is not a merger profile "
+            "(merger-only; fail closed)"
+        )
+
+    if not (session.get("repo") or "").strip():
+        reasons.append("acquired merger lease has no repository")
+
+    pr_number = session.get("pr_number")
+    if not isinstance(pr_number, int) or isinstance(pr_number, bool) or pr_number <= 0:
+        reasons.append("acquired merger lease has no valid PR number")
+
+    if not leases._normalize_sha(session.get("candidate_head")):
+        reasons.append(
+            "acquired merger lease has no pinned candidate_head (exact-head "
+            "scoping required)"
+        )
+
+    return reasons
+
+
 def is_sanctioned_session_lease(session: dict[str, Any] | None) -> bool:
     if not session:
         return False
@@ -131,6 +215,8 @@ def is_sanctioned_session_lease(session: dict[str, Any] | None) -> bool:
         return bool(provenance.get("comment_id")) and bool(
             provenance.get("adopted_from_session_id")
         )
+    if source == SOURCE_ACQUIRE_MERGER:
+        return not assess_acquired_merger_lease_integrity(session)
     if source in {SOURCE_ACQUIRE, SOURCE_HEARTBEAT}:
         return bool(session.get("comment_id") or provenance.get("comment_id"))
     return False
@@ -168,6 +254,8 @@ def describe_session_lease_proof(
     if source == SOURCE_ADOPT and sanctioned:
         kind = "sanctioned_adoption"
         reason = reason or DEFAULT_ADOPTION_REASON
+    elif source == SOURCE_ACQUIRE_MERGER and sanctioned:
+        kind = "sanctioned_acquire_merger"
     elif source == SOURCE_ACQUIRE and sanctioned:
         kind = "sanctioned_acquire"
     elif source == SOURCE_HEARTBEAT and sanctioned:
@@ -216,6 +304,7 @@ _SANCTIONED_LEASE_EVIDENCE_RE = re.compile(
     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_merger|"
     r"lease_proof_kind\s*[:=]\s*sanctioned_acquire|"
     r"adoption_comment_id\s*[:=]|"
     r"sanctioned_adoption|"
@@ -247,6 +336,267 @@ def assess_manual_lease_proof_handoff(report_text: str) -> dict[str, Any]:
     }
 
 
+def format_merger_finalization_body(
+    *,
+    repo: str,
+    pr_number: int,
+    issue_number: int | None,
+    merger_identity: str,
+    merger_profile: str,
+    merger_session_id: str,
+    worktree: str,
+    candidate_head: str | None,
+    target_branch: str,
+    target_branch_sha: str | None,
+    outcome: str,
+    reason: str,
+    lease_comment_id: int | None,
+    finalized_at: datetime | None = None,
+) -> str:
+    """Render the append-only terminal marker for a merger-owned lease (#742).
+
+    The body carries a standard lease marker in a terminal phase, so the
+    existing newest-wins ledger (#577) ends the lease without editing or
+    deleting any prior comment.
+    """
+    finalized_at = finalized_at or datetime.now(timezone.utc)
+    finalized_text = finalized_at.astimezone(timezone.utc).replace(
+        microsecond=0
+    ).isoformat().replace("+00:00", "Z")
+    lease_body = leases.format_lease_body(
+        repo=repo,
+        pr_number=pr_number,
+        issue_number=issue_number,
+        reviewer_identity=merger_identity,
+        profile=merger_profile,
+        session_id=merger_session_id,
+        worktree=worktree,
+        phase=outcome,
+        candidate_head=candidate_head,
+        target_branch=target_branch,
+        target_branch_sha=target_branch_sha,
+        last_activity=finalized_at,
+        blocker=reason,
+    )
+    lines = [
+        MERGER_FINALIZATION_MARKER,
+        f"finalized_at: {finalized_text}",
+        f"finalized_by_identity: {merger_identity}",
+        f"finalized_by_profile: {merger_profile}",
+        f"finalized_by_session_id: {merger_session_id}",
+        f"finalization_outcome: {outcome}",
+        f"finalization_reason: {reason}",
+        f"finalized_lease_comment_id: {lease_comment_id or 'none'}",
+        lease_body,
+    ]
+    return "\n".join(lines)
+
+
+def is_merger_finalization_comment(body: str) -> bool:
+    return MERGER_FINALIZATION_MARKER in (body or "")
+
+
+def assess_merger_lease_finalization(
+    comments: list[dict],
+    *,
+    pr_number: int,
+    session: dict[str, Any] | None,
+    actor_identity: str,
+    actor_profile: str,
+    actor_session_id: str | None,
+    repo: str,
+    worktree: str,
+    candidate_head: str | None,
+    live_head_sha: str | None = None,
+    outcome: str = OUTCOME_RELEASED,
+    reason: str | None = None,
+    runtime_token_fingerprint: str | None = None,
+    issue_number: int | None = None,
+    target_branch: str = "master",
+    target_branch_sha: str | None = None,
+    now: datetime | None = None,
+) -> dict[str, Any]:
+    """Decide whether a merger may terminally finalize its own lease (#742).
+
+    Owner-session only: the caller must hold the exact comment-backed merger
+    lease it is finalizing. Foreign sessions, reviewer profiles, and mismatched
+    repository/PR/head/token-fingerprint callers fail closed. Already-terminal
+    leases return ``already_terminal`` so repeat calls are idempotent and post
+    no second marker.
+    """
+    now = now or datetime.now(timezone.utc)
+    reasons: list[str] = []
+    outcome = (outcome or "").strip().lower()
+    finalization_reason = (reason or "").strip() or DEFAULT_MERGER_FINALIZATION_REASON
+
+    if outcome not in MERGER_FINALIZATION_OUTCOMES:
+        reasons.append(
+            f"outcome '{outcome or 'none'}' is not a terminal merger "
+            f"finalization outcome ({sorted(MERGER_FINALIZATION_OUTCOMES)})"
+        )
+
+    if "merger" not in (actor_profile or "").lower():
+        reasons.append(
+            f"profile '{actor_profile or 'unknown'}' is not a merger profile; "
+            "merger lease finalization is merger-only (fail closed). Reviewer "
+            "sessions use gitea_release_reviewer_pr_lease."
+        )
+
+    session = session or None
+    provenance = (session or {}).get("lease_provenance") or {}
+    if not isinstance(provenance, dict):
+        provenance = {}
+    source = (provenance.get("source") or "").strip()
+
+    if not session:
+        reasons.append(
+            "no in-session merger lease recorded; only the owning session may "
+            "finalize a merger lease"
+        )
+    elif source not in MERGER_OWNED_PROVENANCE_SOURCES:
+        reasons.append(
+            f"in-session lease provenance '{source or 'none'}' is not a "
+            "merger-owned lease; refusing to finalize"
+        )
+    elif not is_sanctioned_session_lease(session):
+        reasons.extend(
+            assess_acquired_merger_lease_integrity(session)
+            if source == SOURCE_ACQUIRE_MERGER
+            else ["in-session merger lease lacks sanctioned provenance"]
+        )
+
+    pinned = leases._normalize_sha(candidate_head)
+    live = leases._normalize_sha(live_head_sha)
+    if not pinned:
+        reasons.append(
+            "candidate_head is required for merger lease finalization "
+            "(exact-head scoping; fail closed)"
+        )
+    if live and pinned and live != pinned:
+        reasons.append(
+            "candidate_head does not match live PR head (fail closed)"
+        )
+
+    if session:
+        session_sid = (session.get("session_id") or "").strip()
+        actor_sid = (actor_session_id or "").strip()
+        if not actor_sid or session_sid != actor_sid:
+            reasons.append(
+                "session_id does not match the in-session merger lease owner; "
+                "foreign-session release is not permitted (fail closed)"
+            )
+        if session.get("pr_number") != pr_number:
+            reasons.append(
+                f"in-session merger lease is for PR #{session.get('pr_number')}, "
+                f"not #{pr_number}"
+            )
+        session_repo = (session.get("repo") or "").strip()
+        if session_repo and session_repo != (repo or "").strip():
+            reasons.append(
+                f"in-session merger lease repository '{session_repo}' does not "
+                f"match '{repo}' (fail closed)"
+            )
+        session_head = leases._normalize_sha(session.get("candidate_head"))
+        if pinned and session_head and session_head != pinned:
+            reasons.append(
+                "in-session merger lease candidate_head does not match the "
+                "supplied candidate_head (fail closed)"
+            )
+        session_identity = (session.get("reviewer_identity") or "").strip()
+        if session_identity and session_identity != (actor_identity or "").strip():
+            reasons.append(
+                "authenticated identity does not match the in-session merger "
+                "lease holder (fail closed)"
+            )
+        session_profile = (session.get("profile") or "").strip()
+        if session_profile and session_profile != (actor_profile or "").strip():
+            reasons.append(
+                "active profile does not match the in-session merger lease "
+                "profile (fail closed)"
+            )
+        recorded_fingerprint = (
+            provenance.get("native_token_fingerprint") or ""
+        ).strip()
+        live_fingerprint = (runtime_token_fingerprint or "").strip()
+        if (
+            recorded_fingerprint
+            and live_fingerprint
+            and recorded_fingerprint != live_fingerprint
+        ):
+            reasons.append(
+                "native runtime token fingerprint does not match the one "
+                "recorded when the merger lease was acquired (fail closed)"
+            )
+
+    lease_comment_id = (session or {}).get("comment_id") or provenance.get("comment_id")
+    entries = leases._lease_entries(comments, pr_number=pr_number)
+    if session and lease_comment_id is not None:
+        if not any(entry.get("comment_id") == lease_comment_id for entry in entries):
+            reasons.append(
+                f"lease marker comment {lease_comment_id} is not present on PR "
+                f"#{pr_number} (fail closed)"
+            )
+
+    active = leases.find_active_reviewer_lease(comments, pr_number=pr_number, now=now)
+    already_terminal = False
+    if active:
+        owner_session = (active.get("session_id") or "").strip()
+        if owner_session and owner_session != (actor_session_id or "").strip():
+            reasons.append(
+                f"active PR lease is owned by session_id={owner_session}; a "
+                "merger may only finalize its own lease (fail closed)"
+            )
+    else:
+        newest = entries[-1] if entries else None
+        newest_phase = ((newest or {}).get("phase") or "").strip().lower()
+        if newest and newest_phase in leases._TERMINAL_PHASES:
+            already_terminal = True
+        elif not entries:
+            reasons.append(
+                f"no comment-backed lease marker found on PR #{pr_number}"
+            )
+
+    finalize_allowed = not reasons and not already_terminal
+    body = None
+    if finalize_allowed and session:
+        body = format_merger_finalization_body(
+            repo=(session.get("repo") or repo),
+            pr_number=pr_number,
+            issue_number=(
+                issue_number
+                if issue_number is not None
+                else session.get("issue_number")
+            ),
+            merger_identity=actor_identity,
+            merger_profile=actor_profile,
+            merger_session_id=(actor_session_id or ""),
+            worktree=worktree or (session.get("worktree") or ""),
+            candidate_head=pinned,
+            target_branch=(
+                session.get("target_branch") or target_branch or "master"
+            ),
+            target_branch_sha=(
+                session.get("target_branch_sha") or target_branch_sha
+            ),
+            outcome=outcome,
+            reason=finalization_reason,
+            lease_comment_id=lease_comment_id,
+            finalized_at=now,
+        )
+
+    return {
+        "finalize_allowed": finalize_allowed,
+        "already_terminal": already_terminal,
+        "reasons": reasons,
+        "outcome": outcome,
+        "finalization_reason": finalization_reason,
+        "active_lease": active,
+        "finalization_body": body,
+        "lease_comment_id": lease_comment_id,
+        "candidate_head": pinned,
+    }
+
+
 def assess_adopt_merger_lease(
     comments: list[dict],
     *,

pr_work_lease.py

@@ -20,7 +20,11 @@ _FIELD_RE = re.compile(
     re.IGNORECASE | re.MULTILINE,
 )
 
-_TERMINAL_REVIEWER_PHASES = frozenset({"done", "released", "blocked"})
+# Must mirror reviewer_pr_lease._TERMINAL_PHASES: both modules read the same
+# append-only lease markers, so a phase that is terminal in one and active in
+# the other yields two conflicting truths for the same comment (#742 review
+# 460). "abandoned" is the owner-session merger finalization outcome.
+_TERMINAL_REVIEWER_PHASES = frozenset({"done", "released", "blocked", "abandoned"})
 _ACTIVE_REVIEWER_PHASES = frozenset({
     "claimed",
     "validating",
@@ -153,25 +157,72 @@ def _lease_phase_active(lease: dict, *, active_phases: frozenset[str]) -> bool:
     ))
 
 
+def _reviewer_chain_key(lease: dict) -> tuple | None:
+    """Identity of the lease chain a reviewer marker belongs to (#742).
+
+    A chain is one session's claim → heartbeat → terminal sequence, keyed by
+    repository, PR, candidate head, session id, identity, and profile. Returns
+    None when any component is missing: an incomplete or malformed marker has
+    no provable chain, so it can neither be cancelled by nor cancel anything.
+    """
+    raw = lease.get("raw_fields") or {}
+    repo = (raw.get("repo") or "").strip().lower()
+    session_id = (lease.get("session_id") or "").strip()
+    identity = (lease.get("reviewer_identity") or "").strip()
+    profile = (lease.get("profile") or "").strip()
+    head = lease.get("candidate_head")
+    pr_number = lease.get("pr_number")
+    if not (repo and session_id and identity and profile and head and pr_number):
+        return None
+    return (repo, pr_number, head, session_id, identity, profile)
+
+
+def _chain_terminated_after(entries: list[dict], index: int) -> bool:
+    """True when a later marker terminates the chain of ``entries[index]``.
+
+    Append-only newest-wins (#577 semantics, chain-scoped for #742): a terminal
+    marker ends only its *own* claim, so a foreign, forged, or malformed
+    terminal marker cannot cancel another session's valid active lease.
+    """
+    key = _reviewer_chain_key(entries[index])
+    if key is None:
+        return False
+    for later in entries[index + 1:]:
+        if (later.get("phase") or "").strip().lower() not in _TERMINAL_REVIEWER_PHASES:
+            continue
+        if _reviewer_chain_key(later) == key:
+            return True
+    return False
+
+
 def find_active_reviewer_lease(
     comments: list[dict],
     *,
     pr_number: int,
     now: datetime | None = None,
 ) -> dict[str, Any] | None:
-    """Return the newest unexpired reviewer lease for *pr_number*, if any."""
+    """Return the newest unexpired, non-terminated reviewer lease for *pr_number*.
+
+    Walking backward past a terminal marker used to resurrect the older claim of
+    the very chain that marker ended, so a released/abandoned finalization still
+    read as an active lease here while ``reviewer_pr_lease`` reported it ended
+    (#742). A claim is now skipped when a later marker terminates its own chain.
+    """
     now = now or datetime.now(timezone.utc)
     candidates = [
         entry for entry in _comment_entries(comments, pr_number=pr_number)
         if entry.get("lease_kind") == "reviewer"
     ]
-    for lease in reversed(candidates):
+    for index in range(len(candidates) - 1, -1, -1):
+        lease = candidates[index]
         if _lease_expired(lease, now=now):
             continue
         phase = (lease.get("phase") or "").strip().lower()
         if phase in _TERMINAL_REVIEWER_PHASES:
             continue
         if phase in _ACTIVE_REVIEWER_PHASES or phase:
+            if _chain_terminated_after(candidates, index):
+                continue
             return lease
     return None
 

reviewer_pr_lease.py

@@ -16,7 +16,10 @@ _FIELD_RE = re.compile(
 )
 _FULL_SHA = re.compile(r"^[0-9a-f]{40}$", re.IGNORECASE)
 
-_TERMINAL_PHASES = frozenset({"done", "released", "blocked"})
+# "abandoned" is the owner-session merger finalization outcome (#742). Without
+# it here, an abandoned marker would fall through to the generic non-empty
+# phase branch and keep re-arming the lease as active.
+_TERMINAL_PHASES = frozenset({"done", "released", "blocked", "abandoned"})
 _ACTIVE_PHASES = frozenset({
     "claimed",
     "validating",

task_capability_map.py

@@ -137,6 +137,17 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
         "permission": "gitea.pr.comment",
         "role": "merger",
     },
+    # #742: owner-session terminal release/abandon of a merger-held lease when
+    # the merge does not occur. Apply path posts an append-only terminal lease
+    # marker (gitea.pr.comment); merger-only, never a reviewer path.
+    "release_merger_pr_lease": {
+        "permission": "gitea.pr.comment",
+        "role": "merger",
+    },
+    "gitea_release_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": {

tests/test_merger_lease_finalization.py

@@ -0,0 +1,956 @@
+import sys as _sys
+from pathlib import Path as _Path
+_sys.path.insert(0, str(_Path(__file__).resolve().parent))
+from mutation_profile_fixture import shared_mutation_env  # noqa: E402,F401
+"""Acquired merger-lease provenance + owner finalization (#742).
+
+Covers the two confirmed defects behind the PR #740 failure:
+
+1. a lease minted by ``gitea_acquire_merger_pr_lease`` was listed as a
+   sanctioned provenance source but rejected by ``is_sanctioned_session_lease``
+   and reported as unsanctioned by ``describe_session_lease_proof``, so the
+   merge gate refused the merger's own freshly acquired lease;
+2. no merger-role owner-session operation existed to terminally release or
+   abandon that lease, leaving natural expiry as the only exit.
+"""
+
+import os
+import sys
+import unittest
+from datetime import datetime, timedelta, timezone
+from unittest.mock import patch
+
+sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
+
+import gitea_mcp_server as mcp_server  # noqa: E402
+import merger_lease_adoption as mla  # noqa: E402
+import pr_work_lease as pwl  # noqa: E402
+import reviewer_pr_lease as leases  # noqa: E402
+import task_capability_map as tcm  # noqa: E402
+
+HEAD = "c" * 40
+OTHER_HEAD = "d" * 40
+REPO = "Scaled-Tech-Consulting/Gitea-Tools"
+PR = 740
+
+
+def _lease_comment(
+    pr_number: int = PR,
+    session_id: str = "merger-session",
+    *,
+    phase: str = "claimed",
+    profile: str = "prgs-merger",
+    identity: str = "merger-user",
+    candidate_head: str = HEAD,
+    comment_id: int = 12354,
+) -> dict:
+    body = leases.format_lease_body(
+        repo=REPO,
+        pr_number=pr_number,
+        issue_number=742,
+        reviewer_identity=identity,
+        profile=profile,
+        session_id=session_id,
+        worktree="branches/merger-pr740",
+        phase=phase,
+        candidate_head=candidate_head,
+        target_branch="master",
+        target_branch_sha="e" * 40,
+        last_activity=datetime.now(timezone.utc),
+    )
+    return {"id": comment_id, "body": body, "user": {"login": identity}}
+
+
+def _acquired_session(**overrides) -> dict:
+    session = {
+        "pr_number": PR,
+        "issue_number": 742,
+        "session_id": "merger-session",
+        "reviewer_identity": "merger-user",
+        "profile": "prgs-merger",
+        "worktree": "branches/merger-pr740",
+        "phase": "claimed",
+        "candidate_head": HEAD,
+        "target_branch": "master",
+        "target_branch_sha": "e" * 40,
+        "repo": REPO,
+        "comment_id": 12354,
+    }
+    session.update(overrides)
+    return session
+
+
+def _record_acquired(session: dict | None = None, **provenance_kwargs) -> dict:
+    provenance = mla.build_lease_provenance(
+        source=mla.SOURCE_ACQUIRE_MERGER,
+        comment_id=provenance_kwargs.pop("comment_id", 12354),
+        **provenance_kwargs,
+    )
+    return leases.record_session_lease(
+        session if session is not None else _acquired_session(),
+        lease_provenance=provenance,
+    )
+
+
+class TestAcquiredMergerProvenance(unittest.TestCase):
+    """AC1/AC2/AC4/AC5/AC6 — provenance sanctioning and proof description."""
+
+    def setUp(self):
+        leases.clear_session_lease()
+
+    def tearDown(self):
+        leases.clear_session_lease()
+
+    def test_acquired_merger_lease_is_sanctioned(self):
+        session = _record_acquired()
+        self.assertTrue(mla.is_sanctioned_session_lease(session))
+        self.assertEqual(mla.assess_acquired_merger_lease_integrity(session), [])
+
+    def test_proof_description_is_explicit_acquired_merger_kind(self):
+        session = _record_acquired()
+        proof = mla.describe_session_lease_proof(session)
+        self.assertEqual(proof["lease_proof_source"], mla.SOURCE_ACQUIRE_MERGER)
+        self.assertEqual(proof["lease_proof_kind"], "sanctioned_acquire_merger")
+        self.assertTrue(proof["lease_proof_sanctioned"])
+        self.assertEqual(proof["lease_proof_comment_id"], 12354)
+
+    def test_manual_session_seed_without_provenance_still_rejected(self):
+        session = leases.record_session_lease(_acquired_session())
+        self.assertFalse(mla.is_sanctioned_session_lease(session))
+        proof = mla.describe_session_lease_proof(session)
+        self.assertEqual(proof["lease_proof_kind"], "unsanctioned_manual_seed")
+
+    def test_unknown_provenance_still_rejected(self):
+        session = leases.record_session_lease(
+            _acquired_session(),
+            lease_provenance={"source": "totally_made_up_tool", "comment_id": 1},
+        )
+        self.assertFalse(mla.is_sanctioned_session_lease(session))
+        self.assertEqual(
+            mla.describe_session_lease_proof(session)["lease_proof_kind"],
+            "unsanctioned",
+        )
+
+    def test_forged_acquired_provenance_without_comment_marker_rejected(self):
+        session = leases.record_session_lease(
+            _acquired_session(comment_id=None),
+            lease_provenance={"source": mla.SOURCE_ACQUIRE_MERGER},
+        )
+        self.assertFalse(mla.is_sanctioned_session_lease(session))
+        self.assertTrue(
+            any(
+                "comment marker" in reason
+                for reason in mla.assess_acquired_merger_lease_integrity(session)
+            )
+        )
+
+    def test_comment_id_mismatch_between_provenance_and_session_rejected(self):
+        session = _record_acquired(_acquired_session(comment_id=999))
+        self.assertFalse(mla.is_sanctioned_session_lease(session))
+
+    def test_reviewer_profile_lease_is_not_acquired_merger_proof(self):
+        session = _record_acquired(_acquired_session(profile="prgs-reviewer"))
+        self.assertFalse(mla.is_sanctioned_session_lease(session))
+        self.assertTrue(
+            any(
+                "not a merger profile" in reason
+                for reason in mla.assess_acquired_merger_lease_integrity(session)
+            )
+        )
+
+    def test_incomplete_fields_fail_closed(self):
+        for field, value in (
+            ("session_id", ""),
+            ("reviewer_identity", ""),
+            ("profile", ""),
+            ("repo", ""),
+            ("pr_number", None),
+            ("candidate_head", None),
+            ("candidate_head", "not-a-sha"),
+        ):
+            with self.subTest(field=field, value=value):
+                leases.clear_session_lease()
+                session = _record_acquired(_acquired_session(**{field: value}))
+                self.assertFalse(
+                    mla.is_sanctioned_session_lease(session),
+                    f"{field}={value!r} must not be sanctioned",
+                )
+
+    def test_existing_reviewer_acquire_and_heartbeat_kinds_unchanged(self):
+        for source, kind in (
+            (mla.SOURCE_ACQUIRE, "sanctioned_acquire"),
+            (mla.SOURCE_HEARTBEAT, "sanctioned_heartbeat"),
+        ):
+            with self.subTest(source=source):
+                leases.clear_session_lease()
+                session = leases.record_session_lease(
+                    {
+                        "pr_number": PR,
+                        "session_id": "reviewer-session",
+                        "candidate_head": HEAD,
+                        "comment_id": 101,
+                    },
+                    lease_provenance=mla.build_lease_provenance(
+                        source=source, comment_id=101
+                    ),
+                )
+                self.assertTrue(mla.is_sanctioned_session_lease(session))
+                self.assertEqual(
+                    mla.describe_session_lease_proof(session)["lease_proof_kind"],
+                    kind,
+                )
+
+    def test_adoption_provenance_unchanged(self):
+        session = leases.record_session_lease(
+            {
+                "pr_number": PR,
+                "session_id": "merger-session",
+                "candidate_head": HEAD,
+                "comment_id": 202,
+            },
+            lease_provenance=mla.build_lease_provenance(
+                source=mla.SOURCE_ADOPT,
+                comment_id=202,
+                adopted_from_session_id="reviewer-session",
+            ),
+        )
+        self.assertTrue(mla.is_sanctioned_session_lease(session))
+        self.assertEqual(
+            mla.describe_session_lease_proof(session)["lease_proof_kind"],
+            "sanctioned_adoption",
+        )
+
+    def test_adoption_without_source_session_still_rejected(self):
+        session = leases.record_session_lease(
+            {"pr_number": PR, "session_id": "merger-session", "comment_id": 202},
+            lease_provenance=mla.build_lease_provenance(
+                source=mla.SOURCE_ADOPT, comment_id=202
+            ),
+        )
+        self.assertFalse(mla.is_sanctioned_session_lease(session))
+
+
+class TestMergeAuthorizationGate(unittest.TestCase):
+    """AC3 — the merge gate accepts a valid freshly acquired merger lease."""
+
+    def setUp(self):
+        leases.clear_session_lease()
+
+    def tearDown(self):
+        leases.clear_session_lease()
+
+    def test_acquired_merger_lease_passes_mutation_lease_gate(self):
+        comments = [_lease_comment()]
+        _record_acquired()
+        result = leases.assess_mutation_lease_gate(
+            pr_number=PR,
+            comments=comments,
+            reviewer_identity="merger-user",
+            session_id="merger-session",
+            mutation="merge",
+            live_head_sha=HEAD,
+            pinned_head_sha=HEAD,
+        )
+        self.assertFalse(result["block"], result["reasons"])
+
+    def test_manual_seed_still_blocked_by_mutation_lease_gate(self):
+        comments = [_lease_comment()]
+        leases.record_session_lease(_acquired_session())
+        result = leases.assess_mutation_lease_gate(
+            pr_number=PR,
+            comments=comments,
+            reviewer_identity="merger-user",
+            session_id="merger-session",
+            mutation="merge",
+            live_head_sha=HEAD,
+            pinned_head_sha=HEAD,
+        )
+        self.assertTrue(result["block"])
+        self.assertTrue(
+            any("sanctioned provenance" in r for r in result["reasons"])
+        )
+
+    def test_head_mismatch_still_blocked_with_acquired_lease(self):
+        comments = [_lease_comment()]
+        _record_acquired()
+        result = leases.assess_mutation_lease_gate(
+            pr_number=PR,
+            comments=comments,
+            reviewer_identity="merger-user",
+            session_id="merger-session",
+            mutation="merge",
+            live_head_sha=OTHER_HEAD,
+            pinned_head_sha=HEAD,
+        )
+        self.assertTrue(result["block"])
+
+
+class TestFinalizationAssessment(unittest.TestCase):
+    """AC7/AC8/AC9/AC10 — owner-only, append-only, idempotent finalization."""
+
+    def setUp(self):
+        leases.clear_session_lease()
+
+    def tearDown(self):
+        leases.clear_session_lease()
+
+    def _assess(self, comments, session, **overrides):
+        kwargs = {
+            "pr_number": PR,
+            "session": session,
+            "actor_identity": "merger-user",
+            "actor_profile": "prgs-merger",
+            "actor_session_id": "merger-session",
+            "repo": REPO,
+            "worktree": "branches/merger-pr740",
+            "candidate_head": HEAD,
+            "live_head_sha": HEAD,
+        }
+        kwargs.update(overrides)
+        return mla.assess_merger_lease_finalization(comments, **kwargs)
+
+    def test_owner_merger_may_release_its_own_acquired_lease(self):
+        session = _record_acquired()
+        result = self._assess([_lease_comment()], session)
+        self.assertTrue(result["finalize_allowed"], result["reasons"])
+        self.assertFalse(result["already_terminal"])
+        self.assertIn(mla.MERGER_FINALIZATION_MARKER, result["finalization_body"])
+        self.assertIn("phase: released", result["finalization_body"])
+
+    def test_abandon_outcome_supported(self):
+        session = _record_acquired()
+        result = self._assess(
+            [_lease_comment()],
+            session,
+            outcome=mla.OUTCOME_ABANDONED,
+            reason="non-retryable-merge-failure",
+        )
+        self.assertTrue(result["finalize_allowed"], result["reasons"])
+        self.assertIn("phase: abandoned", result["finalization_body"])
+        self.assertIn(
+            "finalization_reason: non-retryable-merge-failure",
+            result["finalization_body"],
+        )
+
+    def test_unknown_outcome_rejected(self):
+        session = _record_acquired()
+        result = self._assess([_lease_comment()], session, outcome="deleted")
+        self.assertFalse(result["finalize_allowed"])
+
+    def test_foreign_session_release_rejected(self):
+        session = _record_acquired()
+        comments = [_lease_comment(session_id="someone-else")]
+        result = self._assess(comments, session, actor_session_id="someone-else")
+        self.assertFalse(result["finalize_allowed"])
+        self.assertTrue(
+            any("does not match" in r or "owned by" in r for r in result["reasons"])
+        )
+
+    def test_active_foreign_lease_on_thread_rejected(self):
+        session = _record_acquired()
+        comments = [
+            _lease_comment(),
+            _lease_comment(session_id="other-merger", comment_id=12400),
+        ]
+        result = self._assess(comments, session)
+        self.assertFalse(result["finalize_allowed"])
+        self.assertTrue(any("owned by session_id" in r for r in result["reasons"]))
+
+    def test_reviewer_profile_cannot_finalize_merger_lease(self):
+        session = _record_acquired()
+        result = self._assess(
+            [_lease_comment()], session, actor_profile="prgs-reviewer"
+        )
+        self.assertFalse(result["finalize_allowed"])
+        self.assertTrue(
+            any("not a merger profile" in r for r in result["reasons"])
+        )
+
+    def test_reviewer_sourced_session_lease_is_not_merger_owned(self):
+        session = leases.record_session_lease(
+            _acquired_session(profile="prgs-reviewer"),
+            lease_provenance=mla.build_lease_provenance(
+                source=mla.SOURCE_ACQUIRE, comment_id=12354
+            ),
+        )
+        result = self._assess([_lease_comment(profile="prgs-reviewer")], session)
+        self.assertFalse(result["finalize_allowed"])
+        self.assertTrue(
+            any("merger-owned" in r for r in result["reasons"]), result["reasons"]
+        )
+
+    def test_no_session_lease_rejected(self):
+        result = self._assess([_lease_comment()], None)
+        self.assertFalse(result["finalize_allowed"])
+
+    def test_wrong_pr_rejected(self):
+        session = _record_acquired()
+        result = self._assess(
+            [_lease_comment(pr_number=741, session_id="merger-session")],
+            session,
+            pr_number=741,
+        )
+        self.assertFalse(result["finalize_allowed"])
+
+    def test_wrong_repository_rejected(self):
+        session = _record_acquired()
+        result = self._assess(
+            [_lease_comment()], session, repo="Scaled-Tech-Consulting/Other"
+        )
+        self.assertFalse(result["finalize_allowed"])
+        self.assertTrue(any("repository" in r for r in result["reasons"]))
+
+    def test_wrong_head_rejected(self):
+        session = _record_acquired()
+        result = self._assess(
+            [_lease_comment()], session, candidate_head=OTHER_HEAD
+        )
+        self.assertFalse(result["finalize_allowed"])
+
+    def test_wrong_identity_rejected(self):
+        session = _record_acquired()
+        result = self._assess(
+            [_lease_comment()], session, actor_identity="someone-else"
+        )
+        self.assertFalse(result["finalize_allowed"])
+        self.assertTrue(any("identity" in r for r in result["reasons"]))
+
+    def test_token_fingerprint_mismatch_rejected(self):
+        session = _record_acquired(native_token_fingerprint="fp-acquire")
+        result = self._assess(
+            [_lease_comment()], session, runtime_token_fingerprint="fp-different"
+        )
+        self.assertFalse(result["finalize_allowed"])
+        self.assertTrue(any("fingerprint" in r for r in result["reasons"]))
+
+    def test_matching_token_fingerprint_allowed(self):
+        session = _record_acquired(native_token_fingerprint="fp-acquire")
+        result = self._assess(
+            [_lease_comment()], session, runtime_token_fingerprint="fp-acquire"
+        )
+        self.assertTrue(result["finalize_allowed"], result["reasons"])
+
+    def test_missing_lease_marker_on_thread_rejected(self):
+        session = _record_acquired()
+        result = self._assess([_lease_comment(comment_id=99999)], session)
+        self.assertFalse(result["finalize_allowed"])
+        self.assertTrue(any("lease marker comment" in r for r in result["reasons"]))
+
+    def test_already_terminal_lease_is_idempotent(self):
+        session = _record_acquired()
+        comments = [
+            _lease_comment(),
+            _lease_comment(phase="released", comment_id=12360),
+        ]
+        result = self._assess(comments, session)
+        self.assertTrue(result["already_terminal"])
+        self.assertFalse(result["finalize_allowed"])
+        self.assertIsNone(result["finalization_body"])
+        self.assertEqual(result["reasons"], [])
+
+    def test_abandoned_marker_ends_the_lease(self):
+        comments = [
+            _lease_comment(),
+            _lease_comment(phase="abandoned", comment_id=12361),
+        ]
+        self.assertIsNone(
+            leases.find_active_reviewer_lease(comments, pr_number=PR)
+        )
+
+    def test_finalization_is_append_only(self):
+        session = _record_acquired()
+        original = _lease_comment()
+        comments = [original]
+        result = self._assess(comments, session)
+        self.assertTrue(result["finalize_allowed"], result["reasons"])
+        # The assessment never mutates or removes prior ledger entries.
+        self.assertEqual(comments, [original])
+        self.assertEqual(result["lease_comment_id"], 12354)
+
+
+class TestCrossModuleTerminalPhaseAgreement(unittest.TestCase):
+    """#742 review 460 — reviewer_pr_lease and pr_work_lease must agree.
+
+    Both modules parse the same append-only lease markers. A phase that is
+    terminal in one and active in the other produces two conflicting truths for
+    the same comment, so an abandoned finalization would still read as an active
+    lease to the conflict-fix acquire and PR-sync inventory paths.
+    """
+
+    TERMINAL_PHASES = ("released", "blocked", "done", "abandoned")
+    NONTERMINAL_PHASES = ("claimed", "validating")
+
+    def _both_active(self, phase: str) -> tuple[bool, bool]:
+        comments = [_lease_comment(phase=phase)]
+        return (
+            leases.find_active_reviewer_lease(comments, pr_number=PR) is not None,
+            pwl.find_active_reviewer_lease(comments, pr_number=PR) is not None,
+        )
+
+    def test_abandoned_marker_is_not_active_in_pr_work_lease(self):
+        _, pwl_active = self._both_active("abandoned")
+        self.assertFalse(pwl_active)
+
+    def test_terminal_phases_agree_across_modules(self):
+        for phase in self.TERMINAL_PHASES:
+            with self.subTest(phase=phase):
+                rpl_active, pwl_active = self._both_active(phase)
+                self.assertFalse(rpl_active)
+                self.assertFalse(pwl_active)
+                self.assertEqual(rpl_active, pwl_active)
+
+    def test_nonterminal_phases_remain_active_in_both_modules(self):
+        for phase in self.NONTERMINAL_PHASES:
+            with self.subTest(phase=phase):
+                rpl_active, pwl_active = self._both_active(phase)
+                self.assertTrue(rpl_active)
+                self.assertTrue(pwl_active)
+
+    def test_terminal_phase_sets_are_mirrored(self):
+        self.assertEqual(
+            set(leases._TERMINAL_PHASES), set(pwl._TERMINAL_REVIEWER_PHASES)
+        )
+
+    def test_default_released_finalization_behavior_unchanged(self):
+        # The default outcome is still 'released', and a released marker ends
+        # the lease for both modules exactly as before this change.
+        leases.clear_session_lease()
+        try:
+            session = _record_acquired()
+            result = mla.assess_merger_lease_finalization(
+                [_lease_comment()],
+                pr_number=PR,
+                session=session,
+                actor_identity="merger-user",
+                actor_profile="prgs-merger",
+                actor_session_id="merger-session",
+                repo=REPO,
+                worktree="branches/merger-pr740",
+                candidate_head=HEAD,
+                live_head_sha=HEAD,
+            )
+            self.assertTrue(result["finalize_allowed"], result["reasons"])
+            self.assertEqual(result["outcome"], mla.OUTCOME_RELEASED)
+            self.assertIn("phase: released", result["finalization_body"])
+            self.assertEqual(
+                result["finalization_reason"],
+                mla.DEFAULT_MERGER_FINALIZATION_REASON,
+            )
+        finally:
+            leases.clear_session_lease()
+
+    def test_finalization_appends_and_never_rewrites_prior_markers(self):
+        claimed = _lease_comment()
+        original_body = claimed["body"]
+        ledger = [claimed]
+        # A terminal marker is appended alongside the original claim; the
+        # earlier marker is left byte-for-byte intact and is never removed.
+        terminal = _lease_comment(phase="abandoned", comment_id=12361)
+        ledger.append(terminal)
+        self.assertEqual(len(ledger), 2)
+        self.assertEqual(ledger[0]["body"], original_body)
+        self.assertEqual(ledger[0]["id"], 12354)
+        # Newest-wins (#577): the appended terminal marker ends the lease.
+        self.assertIsNone(leases.find_active_reviewer_lease(ledger, pr_number=PR))
+
+    def test_full_ledger_terminal_phases_end_the_lease_in_both_modules(self):
+        for phase in self.TERMINAL_PHASES:
+            with self.subTest(phase=phase):
+                ledger = [
+                    _lease_comment(),
+                    _lease_comment(phase=phase, comment_id=12361),
+                ]
+                for module in (leases, pwl):
+                    self.assertIsNone(
+                        module.find_active_reviewer_lease(ledger, pr_number=PR),
+                        f"{module.__name__} still active after {phase}",
+                    )
+
+
+class TestFullLedgerNewestWins(unittest.TestCase):
+    """#742 second remediation — chain-scoped newest-wins in pr_work_lease.
+
+    On a realistic append-only ledger (claim -> heartbeat -> terminal),
+    pr_work_lease.find_active_reviewer_lease skipped the terminal marker and
+    walked backward to the older claim of that same chain, resurrecting it. A
+    terminal marker must end its own chain, while a foreign, forged, or
+    malformed terminal marker must never cancel someone else's active lease.
+    """
+
+    OTHER_SESSION = "other-session"
+
+    def _claim(self, **overrides):
+        return _lease_comment(**overrides)
+
+    def _terminal(self, phase="released", comment_id=12361, **overrides):
+        return _lease_comment(phase=phase, comment_id=comment_id, **overrides)
+
+    def _pwl_active(self, ledger):
+        return pwl.find_active_reviewer_lease(ledger, pr_number=PR)
+
+    # --- the chain must end -------------------------------------------------
+
+    def test_claim_then_each_terminal_phase_leaves_no_active_lease(self):
+        for phase in ("released", "blocked", "done", "abandoned"):
+            with self.subTest(phase=phase):
+                self.assertIsNone(
+                    self._pwl_active([self._claim(), self._terminal(phase=phase)])
+                )
+
+    def test_claim_heartbeat_terminal_leaves_no_active_lease(self):
+        for phase in ("released", "blocked", "done", "abandoned"):
+            with self.subTest(phase=phase):
+                ledger = [
+                    self._claim(),
+                    self._lease_validating(),
+                    self._terminal(phase=phase, comment_id=12362),
+                ]
+                self.assertIsNone(self._pwl_active(ledger))
+                self.assertIsNone(
+                    leases.find_active_reviewer_lease(ledger, pr_number=PR)
+                )
+
+    def _lease_validating(self):
+        return _lease_comment(phase="validating", comment_id=12355)
+
+    # --- foreign / forged terminal markers must not cancel ------------------
+
+    def test_foreign_session_terminal_does_not_cancel_active_claim(self):
+        ledger = [
+            self._claim(),
+            self._terminal(session_id=self.OTHER_SESSION),
+        ]
+        active = self._pwl_active(ledger)
+        self.assertIsNotNone(active)
+        self.assertEqual(active["session_id"], "merger-session")
+
+    def test_mismatched_chain_fields_do_not_cancel_active_claim(self):
+        cases = {
+            "identity": {"identity": "someone-else"},
+            "profile": {"profile": "prgs-reviewer"},
+            "candidate_head": {"candidate_head": OTHER_HEAD},
+        }
+        for label, override in cases.items():
+            with self.subTest(field=label):
+                ledger = [self._claim(), self._terminal(**override)]
+                self.assertIsNotNone(
+                    self._pwl_active(ledger),
+                    f"terminal with wrong {label} must not cancel the claim",
+                )
+
+    def test_terminal_for_another_pr_does_not_cancel_active_claim(self):
+        ledger = [
+            self._claim(),
+            self._terminal(pr_number=741, comment_id=12363),
+        ]
+        self.assertIsNotNone(self._pwl_active(ledger))
+
+    def test_terminal_for_another_repository_does_not_cancel_active_claim(self):
+        foreign = self._terminal()
+        foreign["body"] = foreign["body"].replace(
+            REPO, "Scaled-Tech-Consulting/Other-Repo"
+        )
+        self.assertIsNotNone(self._pwl_active([self._claim(), foreign]))
+
+    def test_malformed_terminal_marker_does_not_hide_active_claim(self):
+        malformed = self._terminal()
+        # Strip the session id line: no provable chain, so it cancels nothing.
+        malformed["body"] = "\n".join(
+            line
+            for line in malformed["body"].splitlines()
+            if not line.startswith("session_id:")
+        )
+        active = self._pwl_active([self._claim(), malformed])
+        self.assertIsNotNone(active)
+        self.assertEqual(active["session_id"], "merger-session")
+
+    # --- multi-chain ledgers ------------------------------------------------
+
+    def test_newer_active_chain_survives_older_terminated_chain(self):
+        ledger = [
+            self._claim(),
+            self._terminal(comment_id=12361),
+            _lease_comment(session_id=self.OTHER_SESSION, comment_id=12370),
+        ]
+        active = self._pwl_active(ledger)
+        self.assertIsNotNone(active)
+        self.assertEqual(active["session_id"], self.OTHER_SESSION)
+
+    def test_newest_valid_chain_selected_across_multiple_histories(self):
+        ledger = [
+            _lease_comment(session_id="chain-1", comment_id=12300),
+            _lease_comment(session_id="chain-1", phase="released", comment_id=12301),
+            _lease_comment(session_id="chain-2", comment_id=12310),
+            _lease_comment(session_id="chain-2", phase="abandoned", comment_id=12311),
+            _lease_comment(session_id="chain-3", comment_id=12320),
+        ]
+        active = self._pwl_active(ledger)
+        self.assertIsNotNone(active)
+        self.assertEqual(active["session_id"], "chain-3")
+
+    def test_all_chains_terminated_leaves_no_active_lease(self):
+        ledger = [
+            _lease_comment(session_id="chain-1", comment_id=12300),
+            _lease_comment(session_id="chain-1", phase="released", comment_id=12301),
+            _lease_comment(session_id="chain-2", comment_id=12310),
+            _lease_comment(session_id="chain-2", phase="abandoned", comment_id=12311),
+        ]
+        self.assertIsNone(self._pwl_active(ledger))
+
+    # --- no regression in existing behavior ---------------------------------
+
+    def test_single_marker_behavior_matches_reviewer_pr_lease(self):
+        for phase in (
+            "claimed", "validating", "approved", "merging",
+            "released", "blocked", "done", "abandoned",
+        ):
+            with self.subTest(phase=phase):
+                ledger = [_lease_comment(phase=phase)]
+                self.assertEqual(
+                    pwl.find_active_reviewer_lease(ledger, pr_number=PR) is not None,
+                    leases.find_active_reviewer_lease(ledger, pr_number=PR)
+                    is not None,
+                )
+
+    def test_expired_claim_stays_inactive_and_freshness_unchanged(self):
+        past = datetime.now(timezone.utc) - timedelta(hours=4)
+        expired = {
+            "id": 12399,
+            "user": {"login": "merger-user"},
+            "body": leases.format_lease_body(
+                repo=REPO,
+                pr_number=PR,
+                issue_number=742,
+                reviewer_identity="merger-user",
+                profile="prgs-merger",
+                session_id="expired-session",
+                worktree="branches/merger-pr740",
+                phase="claimed",
+                candidate_head=HEAD,
+                target_branch="master",
+                target_branch_sha="e" * 40,
+                last_activity=past,
+                expires_at=past,
+            ),
+        }
+        self.assertIsNone(self._pwl_active([expired]))
+        self.assertIsNone(
+            leases.find_active_reviewer_lease([expired], pr_number=PR)
+        )
+
+    def test_active_claim_without_any_terminal_marker_remains_active(self):
+        self.assertIsNotNone(self._pwl_active([self._claim()]))
+
+    def test_production_readers_agree_after_release_tool_finalization(self):
+        """The ledger the release tool actually writes must read as terminal."""
+        leases.clear_session_lease()
+        try:
+            session = _record_acquired()
+            result = mla.assess_merger_lease_finalization(
+                [_lease_comment()],
+                pr_number=PR,
+                session=session,
+                actor_identity="merger-user",
+                actor_profile="prgs-merger",
+                actor_session_id="merger-session",
+                repo=REPO,
+                worktree="branches/merger-pr740",
+                candidate_head=HEAD,
+                live_head_sha=HEAD,
+                outcome=mla.OUTCOME_ABANDONED,
+            )
+            self.assertTrue(result["finalize_allowed"], result["reasons"])
+            posted = {
+                "id": 12370,
+                "user": {"login": "merger-user"},
+                "body": result["finalization_body"],
+            }
+            ledger = [_lease_comment(), posted]
+            for module in (leases, pwl):
+                with self.subTest(module=module.__name__):
+                    self.assertIsNone(
+                        module.find_active_reviewer_lease(ledger, pr_number=PR)
+                    )
+            # Append-only: the claim marker is still present and unmodified.
+            self.assertEqual(len(ledger), 2)
+            self.assertEqual(ledger[0]["id"], 12354)
+        finally:
+            leases.clear_session_lease()
+
+
+class TestReleaseMergerLeaseTool(unittest.TestCase):
+    """AC7/AC9/AC10/AC11 — the native tool contract end to end."""
+
+    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 = "release_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
+
+    def _reviewer_profile(self):
+        return {
+            "username": "reviewer-user",
+            "profile_name": "prgs-reviewer",
+            "role": "reviewer",
+            "allowed_operations": [
+                "gitea.read",
+                "gitea.pr.comment",
+                "gitea.pr.review",
+                "gitea.pr.approve",
+            ],
+            "forbidden_operations": ["gitea.pr.merge"],
+        }
+
+    def _call(self, **kwargs):
+        params = {
+            "pr_number": PR,
+            "worktree": "branches/merger-pr740",
+            "candidate_head": HEAD,
+            "remote": "prgs",
+            "org": "Scaled-Tech-Consulting",
+            "repo": "Gitea-Tools",
+        }
+        params.update(kwargs)
+        return mcp_server.gitea_release_merger_pr_lease(**params)
+
+    @patch("gitea_mcp_server.api_request")
+    @patch("gitea_mcp_server._authenticated_username", return_value="merger-user")
+    @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_merger_releases_own_acquired_lease(
+        self, _verify, mock_comments, mock_profile, _resolve, _auth, _user, mock_api
+    ):
+        mock_profile.return_value = self._merger_profile()
+        mock_comments.return_value = [_lease_comment()]
+        mock_api.side_effect = [
+            {"state": "open", "head": {"sha": HEAD}},
+            {"id": 12370},
+        ]
+        _record_acquired()
+        with patch.dict(os.environ, {"GITEA_MCP_PROFILE_NAME": "prgs-merger"}):
+            res = self._call()
+        self.assertTrue(res["success"], res)
+        self.assertTrue(res["finalized"])
+        self.assertEqual(res["finalization_comment_id"], 12370)
+        self.assertEqual(res["outcome"], mla.OUTCOME_RELEASED)
+        # AC11: no active in-session lease survives finalization.
+        self.assertIsNone(leases.get_session_lease())
+
+    @patch("gitea_mcp_server.api_request")
+    @patch("gitea_mcp_server._authenticated_username", return_value="merger-user")
+    @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_merger_cannot_release_foreign_session_lease(
+        self, _verify, mock_comments, mock_profile, _resolve, _auth, _user, mock_api
+    ):
+        mock_profile.return_value = self._merger_profile()
+        mock_comments.return_value = [_lease_comment(session_id="other-merger")]
+        mock_api.side_effect = [{"state": "open", "head": {"sha": HEAD}}]
+        _record_acquired()
+        with patch.dict(os.environ, {"GITEA_MCP_PROFILE_NAME": "prgs-merger"}):
+            res = self._call()
+        self.assertFalse(res["success"], res)
+        self.assertFalse(res["finalized"])
+        self.assertTrue(res["reasons"])
+        # Fail closed: no comment POST was attempted.
+        self.assertEqual(mock_api.call_count, 1)
+        self.assertIsNotNone(leases.get_session_lease())
+
+    @patch("gitea_mcp_server.api_request")
+    @patch("gitea_mcp_server._authenticated_username", return_value="merger-user")
+    @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_already_terminal_release_is_idempotent(
+        self, _verify, mock_comments, mock_profile, _resolve, _auth, _user, mock_api
+    ):
+        mock_profile.return_value = self._merger_profile()
+        mock_comments.return_value = [
+            _lease_comment(),
+            _lease_comment(phase="released", comment_id=12360),
+        ]
+        mock_api.side_effect = [{"state": "open", "head": {"sha": HEAD}}]
+        _record_acquired()
+        with patch.dict(os.environ, {"GITEA_MCP_PROFILE_NAME": "prgs-merger"}):
+            res = self._call()
+        self.assertTrue(res["success"], res)
+        self.assertFalse(res["finalized"])
+        self.assertTrue(res["already_terminal"])
+        # No second marker posted.
+        self.assertEqual(mock_api.call_count, 1)
+        self.assertIsNone(leases.get_session_lease())
+
+    @patch("gitea_mcp_server.get_profile")
+    def test_reviewer_profile_cannot_invoke_merger_release(self, mock_profile):
+        mock_profile.return_value = self._reviewer_profile()
+        _record_acquired()
+        with patch.dict(os.environ, {"GITEA_MCP_PROFILE_NAME": "prgs-reviewer"}):
+            res = self._call()
+        self.assertFalse(res["success"], res)
+        self.assertFalse(res["finalized"])
+        self.assertTrue(res["reasons"])
+        # The lease is untouched by a rejected cross-role call.
+        self.assertIsNotNone(leases.get_session_lease())
+
+
+class TestReleaseMergerLeaseCapability(unittest.TestCase):
+    """AC9 — capability map binds the task to gitea.pr.comment + merger role."""
+
+    def test_task_and_tool_alias_are_merger_scoped(self):
+        for task in ("release_merger_pr_lease", "gitea_release_merger_pr_lease"):
+            with self.subTest(task=task):
+                self.assertEqual(
+                    tcm.required_permission(task), "gitea.pr.comment"
+                )
+                self.assertEqual(tcm.required_role(task), "merger")
+
+    def test_reviewer_release_tool_remains_reviewer_scoped(self):
+        self.assertTrue(hasattr(mcp_server, "gitea_release_reviewer_pr_lease"))
+        self.assertNotEqual(
+            mcp_server.gitea_release_reviewer_pr_lease,
+            mcp_server.gitea_release_merger_pr_lease,
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()