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

Merge pull request 'fix(mcp): accept acquired merger-lease provenance and add owner finalization (Closes #742)' (#743) from fix/issue-742-merger-lease-provenance into master

Browse Source
This commit was merged in pull request #743.
This commit is contained in:
sysadmin
2026-07-18 10:12:27 -05:00
parent a8d2087b4a fde95b9266
commit 0a38d92382
6 changed files with 1567 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gitea_mcp_server.py
+192
View File
@@ -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
+350
View File
@@ -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
+54 -3
View File
@@ -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
+4 -1
View File
@@ -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
+11
View File
@@ -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
+956
View File
@@ -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()