diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 99f355c..b7e5890 100644 --- a/gitea_mcp_server.py +++ b/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", diff --git a/merger_lease_adoption.py b/merger_lease_adoption.py index 46ae926..a8f5eb6 100644 --- a/merger_lease_adoption.py +++ b/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 = "" +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], *, diff --git a/reviewer_pr_lease.py b/reviewer_pr_lease.py index 796a4cb..7541a43 100644 --- a/reviewer_pr_lease.py +++ b/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", diff --git a/task_capability_map.py b/task_capability_map.py index a449d01..83e9dcb 100644 --- a/task_capability_map.py +++ b/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": { diff --git a/tests/test_merger_lease_finalization.py b/tests/test_merger_lease_finalization.py new file mode 100644 index 0000000..7785e08 --- /dev/null +++ b/tests/test_merger_lease_finalization.py @@ -0,0 +1,645 @@ +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, 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 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 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()