Scaled-Tech-Consulting/Gitea-Tools
2
0
Fork 0
Code Issues 73 Pull Requests 9 Actions Packages Projects Releases Wiki Activity

feat: add safe post-merge moot lease cleanup for already-merged PRs (Closes #515)

Browse Source 
Refuse merge-oriented reviewer-lease acquisition/adoption on an already
merged/closed PR, and add a read-first cleanup path for a lingering moot lease.

- reviewer_pr_lease.assess_acquire_lease: new pr_merged_or_closed flag; fail
  closed with a post_merge_moot reason and no lease body when the PR already
  merged/closed.
- reviewer_pr_lease.assess_post_merge_moot_lease: pure assessment. Only treats a
  lease as moot when the PR is merged/closed; never proposes touching an active
  lease on an open PR; provides a terminal phase:released (blocker:
  post-merge-moot) body to neutralise the moot lease append-only; idempotent
  once the released marker is the newest entry.
- gitea_acquire_reviewer_pr_lease: fetch live PR state, pass the flag, surface
  post_merge_moot on refusal (no comment posted).
- gitea_cleanup_post_merge_moot_lease: new read-first tool. Reports merged
  state, merge_commit_sha, linked-issue closure, lease-moot, cleanup
  performed/skipped, and no_merge_or_adoption. apply=True posts the released
  marker only when merged/closed with an active lease; refuses on open PRs.
- tests/test_post_merge_moot_lease.py: 10 tests covering acquire refusal on
  merged PR (no post), safe/idempotent cleanup, and open-PR foreign-lease
  protection.

No relaxation of #407 reviewer-lease or #16 gated-merge gates for open PRs.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
This commit is contained in:
sysadmin
2026-07-08 04:23:17 -04:00
co-authored by Claude Opus 4.8
parent 4d24c34548
commit 52915012a6
3 changed files with 494 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gitea_mcp_server.py
+129
View File
@@ -5064,6 +5064,13 @@ def gitea_acquire_reviewer_pr_lease(
comments = _fetch_pr_comments(
pr_number, remote=remote, host=host, org=org, repo=repo)
# Refuse merge-oriented lease acquisition/adoption on an already-merged or
# closed PR: the lease is moot and adopting it for merge work is unsafe (#515).
pr_live = api_request(
"GET", f"{repo_api_url(h, o, r)}/pulls/{pr_number}", auth) or {}
pr_merged_or_closed = bool(
pr_live.get("merged") or pr_live.get("merged_at")
) or (str(pr_live.get("state") or "").strip().lower() == "closed")
assessment = reviewer_pr_lease.assess_acquire_lease(
comments,
pr_number=pr_number,
@@ -5076,6 +5083,7 @@ def gitea_acquire_reviewer_pr_lease(
candidate_head=candidate_head,
target_branch=target_branch,
target_branch_sha=target_branch_sha,
pr_merged_or_closed=pr_merged_or_closed,
)
if not assessment.get("acquire_allowed"):
return {
@@ -5083,6 +5091,7 @@ def gitea_acquire_reviewer_pr_lease(
"acquired": False,
"reasons": assessment.get("reasons") or [],
"existing_lease": assessment.get("existing_lease"),
"post_merge_moot": assessment.get("post_merge_moot", False),
}
body = assessment["lease_body"]
@@ -5231,6 +5240,126 @@ def gitea_assess_reviewer_pr_lease(
}
@mcp.tool()
def gitea_cleanup_post_merge_moot_lease(
pr_number: int,
apply: bool = False,
remote: str = "dadeschools",
host: str | None = None,
org: str | None = None,
repo: str | None = None,
) -> dict:
"""Safely resolve a moot reviewer lease left on an ALREADY-MERGED/closed PR (#515).
Read-first and fail-safe. This tool never merges and never adopts a lease.
It only acts when the live PR state is merged/closed, and it never steals or
force-cleans an active *foreign* lease while the PR is still open. When
``apply`` is true and a lease is still active on a merged/closed PR, it posts
a terminal ``phase: released`` lease marker (``blocker: post-merge-moot``) —
an append-only comment that neutralises the moot lease without deleting any
other session's comment.
Args:
pr_number: The PR whose lingering lease to assess/clean.
apply: When false (default) report only (read-only). When true, post the
terminal released marker if — and only if — cleanup is allowed.
remote: Known instance — 'dadeschools' or 'prgs'.
host: Override the Gitea host.
org: Override the owner/organization.
repo: Override the repository name.
Returns:
dict reporting PR merged/closed state, merge_commit_sha, linked-issue
closure state, whether the lease is moot, whether cleanup was performed
or skipped (and why), and ``no_merge_or_adoption`` True — this path never
merges or adopts.
"""
read_block = _profile_operation_gate("gitea.read")
if read_block:
return {
"success": False,
"cleanup_performed": False,
"no_merge_or_adoption": True,
"reasons": read_block,
"permission_report": _permission_block_report("gitea.read"),
}
h, o, r = _resolve(remote, host, org, repo)
auth = _auth(h)
pr_live = api_request(
"GET", f"{repo_api_url(h, o, r)}/pulls/{pr_number}", auth) or {}
comments = _fetch_pr_comments(
pr_number, remote=remote, host=host, org=org, repo=repo)
assessment = reviewer_pr_lease.assess_post_merge_moot_lease(
comments,
pr_number=pr_number,
pr_merged=bool(pr_live.get("merged") or pr_live.get("merged_at")),
pr_state=pr_live.get("state"),
merge_commit_sha=pr_live.get("merge_commit_sha"),
)
# Best-effort linked-issue closure state for the report.
active = assessment.get("active_lease") or {}
issue_no = active.get("issue_number")
linked_issue_state = None
if issue_no:
try:
issue = api_request(
"GET", f"{repo_api_url(h, o, r)}/issues/{issue_no}", auth) or {}
linked_issue_state = issue.get("state")
except Exception:
linked_issue_state = None
report = {
"success": True,
"pr_number": pr_number,
"pr_state": assessment.get("pr_state"),
"pr_merged_or_closed": assessment.get("pr_merged_or_closed"),
"merge_commit_sha": assessment.get("merge_commit_sha"),
"linked_issue_number": issue_no,
"linked_issue_state": linked_issue_state,
"lease_moot": assessment.get("is_moot"),
"active_lease": assessment.get("active_lease"),
"cleanup_allowed": assessment.get("cleanup_allowed"),
"cleanup_performed": False,
"no_merge_or_adoption": True,
"mode": "apply" if apply else "read_only",
"reasons": assessment.get("reasons") or [],
}
if not apply:
return report
if not assessment.get("cleanup_allowed"):
report["cleanup_skipped_reason"] = (
assessment.get("reasons") or ["cleanup not allowed"]
)
return report
comment_block = _profile_operation_gate("gitea.pr.comment")
if comment_block:
report["success"] = False
report["reasons"] = comment_block
report["permission_report"] = _permission_block_report("gitea.pr.comment")
return report
verify_preflight_purity(remote)
body = assessment["release_body"]
comment_url = f"{repo_api_url(h, o, r)}/issues/{pr_number}/comments"
with _audited(
"comment_pr",
host=h,
remote=remote,
org=o,
repo=r,
pr_number=pr_number,
request_metadata={"source": "cleanup_post_merge_moot_lease"},
):
posted = api_request("POST", comment_url, auth, {"body": body})
report["cleanup_performed"] = True
report["released_comment_id"] = posted.get("id")
return report
@mcp.tool()
def gitea_list_issue_comments(
issue_number: int,
reviewer_pr_lease.py
+108 -1
View File
@@ -217,12 +217,24 @@ def assess_acquire_lease(
candidate_head: str | None,
target_branch: str,
target_branch_sha: str | None,
pr_merged_or_closed: bool = False,
now: datetime | None = None,
) -> dict[str, Any]:
"""Fail closed when another session holds an active lease."""
"""Fail closed when another session holds an active lease.
When *pr_merged_or_closed* is true the PR has already merged/closed, so any
reviewer-lease acquisition or adoption for merge work is moot: fail closed
with a ``post_merge_moot`` reason and never mint a lease body (#515).
"""
now = now or datetime.now(timezone.utc)
reasons: list[str] = []
existing = find_active_reviewer_lease(comments, pr_number=pr_number, now=now)
post_merge_moot = bool(pr_merged_or_closed)
if post_merge_moot:
reasons.append(
f"post_merge_moot: PR #{pr_number} is already merged/closed; reviewer "
"lease adoption for merge is moot (fail closed)"
)
if existing:
owner_session = (existing.get("session_id") or "").strip()
freshness = existing.get("freshness") or classify_lease_freshness(existing, now=now)
@@ -270,6 +282,101 @@ def assess_acquire_lease(
"existing_lease": existing,
"lease_body": body,
"session_id": session_id,
"post_merge_moot": post_merge_moot,
}
def assess_post_merge_moot_lease(
comments: list[dict],
*,
pr_number: int,
pr_merged: bool = False,
pr_state: str | None = None,
merge_commit_sha: str | None = None,
now: datetime | None = None,
) -> dict[str, Any]:
"""Assess a reviewer lease left lingering on an already-merged/closed PR (#515).
Read-first and fail-safe:
- Only treats a lease as moot when the live PR state is merged/closed.
- Never proposes touching an *active* lease while the PR is still open
(``cleanup_allowed`` stays false and a refusal reason is returned).
- When the PR is merged/closed and a lease is still active, ``cleanup_allowed``
is true and a terminal ``phase: released`` lease body (``blocker:
post-merge-moot``) is provided so the moot lease can be neutralised by an
append-only comment — never by deleting a foreign session's comment, and
never by adopting or merging.
Posting the released body makes that lease terminal, so a subsequent call
finds no active lease and reports nothing left to clean (idempotent).
"""
now = now or datetime.now(timezone.utc)
merged_or_closed = bool(pr_merged) or (
str(pr_state or "").strip().lower() == "closed"
)
active = find_active_reviewer_lease(comments, pr_number=pr_number, now=now)
# The newest lease comment is authoritative: once a terminal marker
# (released/done/blocked) is the latest entry, the lease is resolved even if
# an earlier non-terminal comment from the same session still lingers. This
# keeps post-merge cleanup idempotent.
entries = _lease_entries(comments, pr_number=pr_number)
newest = entries[-1] if entries else None
newest_terminal = bool(newest) and (
(newest.get("phase") or "").strip().lower() in _TERMINAL_PHASES
)
reasons: list[str] = []
cleanup_allowed = False
release_body: str | None = None
is_moot = bool(active) and merged_or_closed and not newest_terminal
if not merged_or_closed:
if active:
reasons.append(
f"PR #{pr_number} is still open; refusing to touch active reviewer "
"lease (fail closed)"
)
else:
reasons.append(
f"PR #{pr_number} is still open; no post-merge lease cleanup applicable"
)
elif newest_terminal:
reasons.append(
f"PR #{pr_number} reviewer lease already released/terminal; nothing to clean"
)
elif active:
cleanup_allowed = True
release_body = format_lease_body(
repo=active.get("repo") or "",
pr_number=pr_number,
issue_number=active.get("issue_number"),
reviewer_identity=active.get("reviewer_identity") or "",
profile=active.get("profile") or "unknown",
session_id=active.get("session_id") or "",
worktree=active.get("worktree") or "",
phase="released",
candidate_head=active.get("candidate_head"),
target_branch=active.get("target_branch") or "master",
target_branch_sha=active.get("target_branch_sha"),
last_activity=now,
blocker="post-merge-moot",
)
else:
reasons.append(
f"PR #{pr_number} is merged/closed but no active reviewer lease remains; "
"nothing to clean"
)
return {
"pr_number": pr_number,
"pr_state": pr_state,
"pr_merged_or_closed": merged_or_closed,
"merge_commit_sha": merge_commit_sha,
"active_lease": active,
"is_moot": is_moot,
"cleanup_allowed": cleanup_allowed,
"release_body": release_body,
"reasons": reasons,
}
tests/test_post_merge_moot_lease.py
+257
View File
@@ -0,0 +1,257 @@
"""Post-merge moot reviewer-lease handling (#515).
Covers:
a. Reviewer-lease acquisition/adoption is refused on an already-merged/closed
PR (fail closed, no mutation).
b. The post-merge moot cleanup path is safe and idempotent.
c. An active foreign lease on an *open* PR is never force-cleaned.
"""
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 reviewer_pr_lease as leases # noqa: E402
from mcp_server import ( # noqa: E402
gitea_acquire_reviewer_pr_lease,
gitea_cleanup_post_merge_moot_lease,
)
FAKE_AUTH = "Basic dGVzdDp0ZXN0"
MERGER_ENV = {
"GITEA_PROFILE_NAME": "prgs-merger",
"GITEA_ALLOWED_OPERATIONS": "gitea.read,gitea.pr.comment",
}
PR = 487
ISSUE = 485
SESSION = "97274-676d20a825c4"
def _lease_comment(pr_number=PR, session_id=SESSION, *, phase="claimed",
candidate_head="a" * 40):
body = leases.format_lease_body(
repo="Scaled-Tech-Consulting/Gitea-Tools",
pr_number=pr_number,
issue_number=ISSUE,
reviewer_identity="sysadmin",
profile="prgs-reviewer",
session_id=session_id,
worktree="branches/review-pr487",
phase=phase,
candidate_head=candidate_head,
target_branch="master",
target_branch_sha="b" * 40,
last_activity=datetime.now(timezone.utc),
)
return {"id": 6603, "body": body, "user": {"login": "sysadmin"}}
# --------------------------------------------------------------------------- #
# Pure logic
# --------------------------------------------------------------------------- #
class TestAcquireRefusedOnMergedPR(unittest.TestCase):
def test_acquire_refused_when_pr_merged_or_closed(self):
result = leases.assess_acquire_lease(
[_lease_comment()],
pr_number=PR,
reviewer_identity="sysadmin",
profile="prgs-merger",
session_id="new-session",
repo="Scaled-Tech-Consulting/Gitea-Tools",
issue_number=ISSUE,
worktree="branches/merge-pr487",
candidate_head="a" * 40,
target_branch="master",
target_branch_sha="b" * 40,
pr_merged_or_closed=True,
)
self.assertFalse(result["acquire_allowed"])
self.assertTrue(result["post_merge_moot"])
self.assertIsNone(result["lease_body"])
self.assertTrue(any("post_merge_moot" in r for r in result["reasons"]))
def test_acquire_still_allowed_on_open_pr_without_flag(self):
result = leases.assess_acquire_lease(
[],
pr_number=PR,
reviewer_identity="sysadmin",
profile="prgs-reviewer",
session_id="s1",
repo="Scaled-Tech-Consulting/Gitea-Tools",
issue_number=ISSUE,
worktree="branches/review-pr487",
candidate_head="a" * 40,
target_branch="master",
target_branch_sha="b" * 40,
)
self.assertTrue(result["acquire_allowed"])
self.assertFalse(result["post_merge_moot"])
class TestPostMergeMootAssessment(unittest.TestCase):
def test_merged_pr_with_active_lease_is_moot_and_cleanable(self):
a = leases.assess_post_merge_moot_lease(
[_lease_comment()],
pr_number=PR,
pr_merged=True,
pr_state="closed",
merge_commit_sha="c" * 40,
)
self.assertTrue(a["pr_merged_or_closed"])
self.assertTrue(a["is_moot"])
self.assertTrue(a["cleanup_allowed"])
self.assertIsNotNone(a["release_body"])
self.assertIn("phase: released", a["release_body"])
self.assertIn("blocker: post-merge-moot", a["release_body"])
def test_open_pr_active_lease_never_cleaned(self):
a = leases.assess_post_merge_moot_lease(
[_lease_comment()],
pr_number=PR,
pr_merged=False,
pr_state="open",
)
self.assertFalse(a["pr_merged_or_closed"])
self.assertFalse(a["is_moot"])
self.assertFalse(a["cleanup_allowed"])
self.assertIsNone(a["release_body"])
self.assertTrue(any("still open" in r for r in a["reasons"]))
def test_merged_pr_without_lease_nothing_to_clean(self):
a = leases.assess_post_merge_moot_lease(
[], pr_number=PR, pr_merged=True, pr_state="closed")
self.assertTrue(a["pr_merged_or_closed"])
self.assertFalse(a["is_moot"])
self.assertFalse(a["cleanup_allowed"])
self.assertTrue(any("nothing to clean" in r for r in a["reasons"]))
def test_cleanup_is_idempotent(self):
"""After the released marker is posted, a re-assess finds nothing to clean."""
first = leases.assess_post_merge_moot_lease(
[_lease_comment()], pr_number=PR, pr_merged=True, pr_state="closed")
self.assertTrue(first["cleanup_allowed"])
released_comment = {
"id": 7000, "body": first["release_body"], "user": {"login": "sysadmin"}}
second = leases.assess_post_merge_moot_lease(
[_lease_comment(), released_comment],
pr_number=PR, pr_merged=True, pr_state="closed")
self.assertFalse(second["is_moot"])
self.assertFalse(second["cleanup_allowed"])
# --------------------------------------------------------------------------- #
# Server tools
# --------------------------------------------------------------------------- #
def _api_side_effect(*, pr_state, pr_merged, comments, posted_id=9999):
"""Build an api_request side effect keyed on method + url."""
calls = {"post": []}
def _side(method, url, auth=None, payload=None, *a, **k):
m = (method or "").upper()
if m == "POST":
calls["post"].append({"url": url, "payload": payload})
return {"id": posted_id}
if "/comments" in url:
return list(comments)
if "/pulls/" in url:
pr = {"state": pr_state, "number": PR, "merge_commit_sha": "c" * 40}
if pr_merged:
pr["merged"] = True
pr["merged_at"] = "2026-07-08T07:46:04Z"
return pr
if "/issues/" in url:
return {"state": "closed" if pr_merged else "open", "number": ISSUE}
return {}
return _side, calls
class TestAcquireToolRefusesMergedPR(unittest.TestCase):
def setUp(self):
leases.clear_session_lease()
@patch("mcp_server.verify_preflight_purity", return_value=None)
@patch("mcp_server._authenticated_username", return_value="sysadmin")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
@patch("mcp_server.api_request")
def test_acquire_tool_fails_closed_on_merged_pr_without_posting(
self, mock_api, _auth, _user, _purity):
side, calls = _api_side_effect(
pr_state="closed", pr_merged=True, comments=[])
mock_api.side_effect = side
with patch.dict(os.environ, MERGER_ENV, clear=True):
result = gitea_acquire_reviewer_pr_lease(
pr_number=PR,
worktree="branches/merge-pr487",
candidate_head="a" * 40,
issue_number=ISSUE,
remote="prgs",
)
self.assertFalse(result["success"])
self.assertFalse(result["acquired"])
self.assertTrue(result.get("post_merge_moot"))
self.assertEqual(calls["post"], [], "must not post a lease comment")
class TestCleanupTool(unittest.TestCase):
def setUp(self):
leases.clear_session_lease()
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
@patch("mcp_server.api_request")
def test_read_only_reports_moot_without_mutating(self, mock_api, _auth):
side, calls = _api_side_effect(
pr_state="closed", pr_merged=True, comments=[_lease_comment()])
mock_api.side_effect = side
with patch.dict(os.environ, MERGER_ENV, clear=True):
result = gitea_cleanup_post_merge_moot_lease(
pr_number=PR, apply=False, remote="prgs")
self.assertTrue(result["success"])
self.assertTrue(result["pr_merged_or_closed"])
self.assertTrue(result["lease_moot"])
self.assertFalse(result["cleanup_performed"])
self.assertTrue(result["no_merge_or_adoption"])
self.assertEqual(result["mode"], "read_only")
self.assertEqual(calls["post"], [])
@patch("mcp_server.verify_preflight_purity", return_value=None)
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
@patch("mcp_server.api_request")
def test_apply_posts_released_marker_on_merged_pr(
self, mock_api, _auth, _purity):
side, calls = _api_side_effect(
pr_state="closed", pr_merged=True, comments=[_lease_comment()])
mock_api.side_effect = side
with patch.dict(os.environ, MERGER_ENV, clear=True):
result = gitea_cleanup_post_merge_moot_lease(
pr_number=PR, apply=True, remote="prgs")
self.assertTrue(result["success"])
self.assertTrue(result["cleanup_performed"])
self.assertEqual(result["released_comment_id"], 9999)
self.assertEqual(len(calls["post"]), 1)
self.assertIn("phase: released", calls["post"][0]["payload"]["body"])
self.assertIn("post-merge-moot", calls["post"][0]["payload"]["body"])
@patch("mcp_server.verify_preflight_purity", return_value=None)
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
@patch("mcp_server.api_request")
def test_apply_refuses_to_clean_open_pr(self, mock_api, _auth, _purity):
side, calls = _api_side_effect(
pr_state="open", pr_merged=False, comments=[_lease_comment()])
mock_api.side_effect = side
with patch.dict(os.environ, MERGER_ENV, clear=True):
result = gitea_cleanup_post_merge_moot_lease(
pr_number=PR, apply=True, remote="prgs")
self.assertFalse(result["cleanup_performed"])
self.assertFalse(result["pr_merged_or_closed"])
self.assertEqual(calls["post"], [], "never force-clean an open PR lease")
self.assertTrue(
any("still open" in r for r in result.get("cleanup_skipped_reason", [])))
if __name__ == "__main__":
unittest.main()