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

Merge pull request 'feat: add conflict-fix leases and stale-head protection (Closes #399)' (#416) from feat/issue-399-conflict-fix-leases into master

Browse Source
This commit was merged in pull request #416.
This commit is contained in:
sysadmin
2026-07-08 00:34:20 -05:00
parent cc5686b896 d4dc9aa854
commit ae0478bf54
14 changed files with 1246 additions and 99 deletions
Download Patch File Download Diff File Expand all files Collapse all files
final_report_validator.py
+41
View File
@@ -491,6 +491,45 @@ def _rule_reviewer_validation_failure_history(
]
def _rule_reviewer_stale_head_proof(report_text: str) -> list[dict[str, str]]:
from pr_work_lease import assess_reviewer_stale_head_final_report
result = assess_reviewer_stale_head_final_report(report_text)
if result.get("proven"):
return []
return _findings_from_reasons(
"reviewer.stale_head_proof",
result.get("reasons") or [],
field="Stale-head proof",
severity="block",
safe_next_action=(
"state reviewed head SHA, live head before approval/merge, and "
"whether any push occurred during validation"
),
)
def _rule_conflict_fix_push_proof(report_text: str) -> list[dict[str, str]]:
from pr_work_lease import assess_conflict_fix_final_report
text = report_text or ""
if "conflict-fix" not in text.lower() and "conflict fix" not in text.lower():
return []
result = assess_conflict_fix_final_report(text)
if result.get("proven"):
return []
return _findings_from_reasons(
"author.conflict_fix_push_proof",
result.get("reasons") or [],
field="Conflict-fix push proof",
severity="block",
safe_next_action=(
"state branch head before/after push, reviewer lease status, "
"fast-forward status, and whether any reviewer was active"
),
)
def _rule_reviewer_validation_command(report_text: str) -> list[dict[str, str]]:
text = report_text or ""
if not _BARE_PYTEST_RE.search(text):
@@ -965,6 +1004,7 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
_rule_reviewer_target_branch_freshness,
_rule_reviewer_mutation_ledger,
_rule_reviewer_review_mutation,
_rule_reviewer_stale_head_proof,
],
"reconcile_already_landed": [
_rule_reconcile_controller_handoff,
@@ -989,6 +1029,7 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
_rule_shared_email_disclosure,
*_SHARED_ISSUE_LOCK_RULES,
_rule_reviewer_vague_mutations_none,
_rule_conflict_fix_push_proof,
],
"issue_filing": [
_rule_shared_controller_handoff,
gitea_mcp_server.py
+222 -1
View File
@@ -545,6 +545,7 @@ import merged_cleanup_reconcile # noqa: E402
import reconciler_profile # noqa: E402
import reconciliation_workflow # noqa: E402
import review_merge_state_machine # noqa: E402
import pr_work_lease # noqa: E402
import native_mcp_preference # noqa: E402
@@ -2501,6 +2502,50 @@ def gitea_get_pr_review_feedback(
}
def _list_pr_lease_comments(
pr_number: int,
*,
remote: str,
host: str | None,
org: str | None,
repo: str | None,
limit: int = 100,
) -> list[dict]:
"""Fetch PR/issue thread comments used for reviewer/conflict-fix leases."""
h, o, r = _resolve(remote, host, org, repo)
auth = _auth(h)
api = f"{repo_api_url(h, o, r)}/issues/{pr_number}/comments"
comments = api_request("GET", api, auth) or []
return list(comments[:limit])
def _pr_work_lease_reviewer_block(
*,
pr_number: int,
reviewed_head_sha: str | None,
live_head_sha: str | None,
mutation: str,
remote: str,
host: str | None,
org: str | None,
repo: str | None,
) -> dict:
comments = _list_pr_lease_comments(
pr_number,
remote=remote,
host=host,
org=org,
repo=repo,
)
return pr_work_lease.assess_reviewer_mutation_blocked(
pr_number=pr_number,
comments=comments,
reviewed_head_sha=reviewed_head_sha,
live_head_sha=live_head_sha,
mutation=mutation,
)
def _evaluate_pr_review_submission(
pr_number: int,
action: str,
@@ -2599,6 +2644,11 @@ def _evaluate_pr_review_submission(
lock = _load_review_decision_lock() or {}
if live and lock.get("ready_expected_head_sha"):
pinned_sha = lock.get("ready_expected_head_sha")
if live and not pinned_sha:
reasons.append(
"reviewed head SHA required before live review mutation (fail closed, #399)"
)
return result
if pinned_sha and actual_sha and pinned_sha != actual_sha:
reasons.append(
"expected head SHA does not match current PR head (fail closed)"
@@ -2608,6 +2658,21 @@ def _evaluate_pr_review_submission(
reasons.append("PR head SHA unavailable (fail closed)")
return result
lease_block = _pr_work_lease_reviewer_block(
pr_number=pr_number,
reviewed_head_sha=pinned_sha,
live_head_sha=actual_sha,
mutation=action,
remote=remote,
host=host,
org=org,
repo=repo,
)
if lease_block.get("block"):
reasons.extend(lease_block.get("reasons") or [])
result["pr_work_lease"] = lease_block
return result
result["would_perform"] = True
if not live:
reasons.append(
@@ -2750,6 +2815,39 @@ def gitea_mark_final_review_decision(
f"{sorted(_REVIEW_ACTIONS)}"
],
}
if not (expected_head_sha or "").strip():
return {
"marked_ready": False,
"reasons": [
"expected_head_sha required before marking final review "
"decision (fail closed, #399)"
],
}
elig = gitea_check_pr_eligibility(
pr_number=pr_number,
action="review",
remote=remote,
host=None,
org=org,
repo=repo,
)
live_head = elig.get("head_sha")
lease_block = _pr_work_lease_reviewer_block(
pr_number=pr_number,
reviewed_head_sha=expected_head_sha,
live_head_sha=live_head,
mutation="mark_ready",
remote=remote,
host=None,
org=org,
repo=repo,
)
if lease_block.get("block"):
return {
"marked_ready": False,
"reasons": lease_block.get("reasons") or [],
"pr_work_lease": lease_block,
}
if action == "request_changes":
# Duplicate request-changes suppression (#332): an unresolved
# REQUEST_CHANGES at the current head must not be duplicated.
@@ -3394,8 +3492,27 @@ def gitea_merge_pr(
if reasons:
return result
# Gate 4 — head SHA must match if the caller pinned a reviewed SHA.
# Gate 4 — reviewed head SHA is mandatory and must match live PR head (#399).
actual_sha = result["head_sha"]
if not (expected_head_sha or "").strip():
reasons.append(
"expected_head_sha required before merge (fail closed, #399)"
)
return result
lease_block = _pr_work_lease_reviewer_block(
pr_number=pr_number,
reviewed_head_sha=expected_head_sha,
live_head_sha=actual_sha,
mutation="merge",
remote=remote,
host=host,
org=org,
repo=repo,
)
if lease_block.get("block"):
reasons.extend(lease_block.get("reasons") or [])
result["pr_work_lease"] = lease_block
return result
if expected_head_sha and actual_sha and expected_head_sha != actual_sha:
reasons.append(
"expected head SHA does not match current PR head (fail closed)"
@@ -6691,6 +6808,110 @@ def gitea_post_heartbeat(
)
@mcp.tool()
def gitea_acquire_conflict_fix_lease(
pr_number: int,
branch: str,
worktree_path: str,
head_before: str,
remote: str = "dadeschools",
host: str | None = None,
org: str | None = None,
repo: str | None = None,
) -> dict:
"""Acquire a conflict-fix lease on a PR branch before pushing (#399)."""
blocked = _profile_permission_block(
task_capability_map.required_permission("comment_issue"))
if blocked:
return blocked
verify_preflight_purity(remote, worktree_path=worktree_path)
comments = _list_pr_lease_comments(
pr_number,
remote=remote,
host=host,
org=org,
repo=repo,
)
reviewer_lease = pr_work_lease.find_active_reviewer_lease(
comments, pr_number=pr_number)
if reviewer_lease:
return {
"acquired": False,
"reasons": [
f"active reviewer lease on PR #{pr_number}; cannot acquire "
"conflict-fix lease (fail closed)"
],
"active_reviewer_lease": reviewer_lease,
}
profile_name = get_profile().get("profile_name") or "unknown"
body = pr_work_lease.format_conflict_fix_lease_body(
pr_number=pr_number,
branch=branch,
worktree=worktree_path,
profile=profile_name,
head_before=head_before,
reviewer_active=bool(reviewer_lease),
)
posted = _post_structured_issue_comment(
issue_number=pr_number,
body=body,
remote=remote,
host=host,
org=org,
repo=repo,
audit_op="conflict_fix_lease_acquire",
)
return {
"acquired": posted.get("success", False),
"pr_number": pr_number,
"branch": branch,
"worktree_path": worktree_path,
"head_before": head_before,
"comment_id": posted.get("comment_id"),
"active_reviewer_lease": reviewer_lease,
"reasons": [] if posted.get("success") else ["lease comment post failed"],
}
@mcp.tool()
def gitea_assess_conflict_fix_push(
pr_number: int,
branch_head_before: str,
branch_head_after: str,
worktree_path: str,
push_cwd: str,
is_fast_forward: bool = True,
remote: str = "dadeschools",
host: str | None = None,
org: str | None = None,
repo: str | None = None,
) -> dict:
"""Read-only pre-push gate for author conflict-fix sessions (#399)."""
read_block = _profile_operation_gate("gitea.read")
if read_block:
return {
"push_allowed": False,
"reasons": read_block,
"permission_report": _permission_block_report("gitea.read"),
}
comments = _list_pr_lease_comments(
pr_number,
remote=remote,
host=host,
org=org,
repo=repo,
)
return pr_work_lease.assess_conflict_fix_push(
pr_number=pr_number,
comments=comments,
branch_head_before=branch_head_before,
branch_head_after=branch_head_after,
worktree_path=worktree_path,
push_cwd=push_cwd,
is_fast_forward=is_fast_forward,
)
@mcp.tool()
def gitea_reconcile_issue_claims(
state: str = "open",
pr_work_lease.py
+482
View File
@@ -0,0 +1,482 @@
"""Conflict-fix and reviewer PR work leases (#399, #407 reader).
Structured PR/issue comments prove exclusive phases so author conflict-fix
pushes cannot race reviewer validation/approval/merge on the same head.
"""
from __future__ import annotations
import re
from datetime import datetime, timedelta, timezone
from typing import Any
REVIEWER_LEASE_MARKER = "<!-- mcp-review-lease:v1 -->"
CONFLICT_FIX_LEASE_MARKER = "<!-- mcp-conflict-fix-lease:v1 -->"
_FULL_SHA = re.compile(r"^[0-9a-f]{40}$", re.IGNORECASE)
_FIELD_RE = re.compile(
r"^\s*([a-z_]+)\s*:\s*(.+?)\s*$",
re.IGNORECASE | re.MULTILINE,
)
_TERMINAL_REVIEWER_PHASES = frozenset({"done", "released", "blocked"})
_ACTIVE_REVIEWER_PHASES = frozenset({
"claimed",
"validating",
"approved",
"request-changes",
"merging",
})
_TERMINAL_CONFLICT_FIX_PHASES = frozenset({"released", "blocked", "done"})
_ACTIVE_CONFLICT_FIX_PHASES = frozenset({"claimed", "pushing", "pushed"})
DEFAULT_CONFLICT_FIX_TTL_MINUTES = 120
DEFAULT_REVIEWER_LEASE_TTL_MINUTES = 120
def _parse_timestamp(value: str | None) -> datetime | None:
if not value:
return None
text = value.strip()
if text.endswith("Z"):
text = text[:-1] + "+00:00"
try:
parsed = datetime.fromisoformat(text)
except ValueError:
return None
if parsed.tzinfo is None:
return parsed.replace(tzinfo=timezone.utc)
return parsed.astimezone(timezone.utc)
def _normalize_sha(value: str | None) -> str | None:
text = (value or "").strip().lower()
if not text:
return None
return text if _FULL_SHA.match(text) else None
def _parse_pr_ref(value: str | None) -> int | None:
digits = re.sub(r"[^\d]", "", value or "")
return int(digits) if digits.isdigit() else None
def _parse_marker_comment(body: str, marker: str) -> dict[str, str] | None:
text = body or ""
if marker not in text:
return None
fields: dict[str, str] = {}
for match in _FIELD_RE.finditer(text):
fields[match.group(1).strip().lower()] = match.group(2).strip()
return fields or None
def parse_reviewer_lease_comment(body: str) -> dict[str, Any] | None:
fields = _parse_marker_comment(body, REVIEWER_LEASE_MARKER)
if not fields:
return None
return {
"lease_kind": "reviewer",
"pr_number": _parse_pr_ref(fields.get("pr")),
"issue_number": _parse_pr_ref(fields.get("issue")),
"reviewer_identity": fields.get("reviewer_identity"),
"profile": fields.get("profile"),
"session_id": fields.get("session_id"),
"worktree": fields.get("worktree"),
"phase": (fields.get("phase") or "").strip().lower() or None,
"candidate_head": _normalize_sha(fields.get("candidate_head")),
"target_branch": fields.get("target_branch"),
"target_branch_sha": _normalize_sha(fields.get("target_branch_sha")),
"last_activity": fields.get("last_activity"),
"expires_at": fields.get("expires_at"),
"blocker": fields.get("blocker"),
"raw_fields": fields,
}
def parse_conflict_fix_lease_comment(body: str) -> dict[str, Any] | None:
fields = _parse_marker_comment(body, CONFLICT_FIX_LEASE_MARKER)
if not fields:
return None
ff = (fields.get("fast_forward") or "").strip().lower()
reviewer_active = (fields.get("reviewer_active") or "").strip().lower()
return {
"lease_kind": "conflict_fix",
"pr_number": _parse_pr_ref(fields.get("pr")),
"branch": fields.get("branch"),
"worktree": fields.get("worktree"),
"profile": fields.get("profile"),
"session_id": fields.get("session_id"),
"phase": (fields.get("phase") or "").strip().lower() or None,
"head_before": _normalize_sha(fields.get("head_before")),
"head_after": _normalize_sha(fields.get("head_after")),
"expires_at": fields.get("expires_at"),
"reviewer_active": reviewer_active in {"yes", "true", "1"},
"fast_forward": ff in {"yes", "true", "1"},
"raw_fields": fields,
}
def _comment_entries(comments: list[dict], *, pr_number: int | None) -> list[dict]:
entries: list[dict] = []
for comment in comments or []:
body = comment.get("body") or ""
for parser in (parse_reviewer_lease_comment, parse_conflict_fix_lease_comment):
parsed = parser(body)
if not parsed:
continue
if pr_number is not None and parsed.get("pr_number") not in (None, pr_number):
continue
entries.append({
**parsed,
"comment_id": comment.get("id"),
"author": (comment.get("user") or {}).get("login") or comment.get("author"),
"created_at": comment.get("created_at"),
"updated_at": comment.get("updated_at"),
})
break
return entries
def _lease_expired(lease: dict, *, now: datetime) -> bool:
expires_at = _parse_timestamp(lease.get("expires_at"))
return bool(expires_at and expires_at <= now)
def _lease_phase_active(lease: dict, *, active_phases: frozenset[str]) -> bool:
phase = (lease.get("phase") or "").strip().lower()
if phase in _TERMINAL_REVIEWER_PHASES or phase in _TERMINAL_CONFLICT_FIX_PHASES:
return False
return phase in active_phases or bool(phase and phase not in (
_TERMINAL_REVIEWER_PHASES | _TERMINAL_CONFLICT_FIX_PHASES
))
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."""
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):
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:
return lease
return None
def find_active_conflict_fix_lease(
comments: list[dict],
*,
pr_number: int,
now: datetime | None = None,
) -> dict[str, Any] | None:
"""Return the newest unexpired conflict-fix lease for *pr_number*, if any."""
now = now or datetime.now(timezone.utc)
candidates = [
entry for entry in _comment_entries(comments, pr_number=pr_number)
if entry.get("lease_kind") == "conflict_fix"
]
for lease in reversed(candidates):
if _lease_expired(lease, now=now):
continue
phase = (lease.get("phase") or "").strip().lower()
if phase in _TERMINAL_CONFLICT_FIX_PHASES:
continue
if phase in _ACTIVE_CONFLICT_FIX_PHASES or phase:
return lease
return None
def format_conflict_fix_lease_body(
*,
pr_number: int,
branch: str,
worktree: str,
profile: str,
head_before: str,
phase: str = "claimed",
session_id: str = "unknown",
expires_at: datetime | None = None,
reviewer_active: bool = False,
) -> str:
expires = expires_at or (
datetime.now(timezone.utc) + timedelta(minutes=DEFAULT_CONFLICT_FIX_TTL_MINUTES)
)
expires_text = expires.astimezone(timezone.utc).replace(microsecond=0).isoformat().replace(
"+00:00", "Z"
)
lines = [
CONFLICT_FIX_LEASE_MARKER,
f"pr: #{pr_number}",
f"branch: {branch}",
f"worktree: {worktree}",
f"profile: {profile}",
f"session_id: {session_id}",
f"phase: {phase}",
f"head_before: {head_before}",
f"expires_at: {expires_text}",
f"reviewer_active: {'yes' if reviewer_active else 'no'}",
]
return "\n".join(lines)
def assess_head_sha_equality(
reviewed_head_sha: str | None,
live_head_sha: str | None,
) -> dict[str, Any]:
"""Fail closed when reviewed and live PR heads differ."""
reviewed = _normalize_sha(reviewed_head_sha)
live = _normalize_sha(live_head_sha)
reasons: list[str] = []
if not reviewed or not live:
reasons.append(
"reviewed/live head SHA missing or not full 40-hex; fail closed"
)
elif reviewed != live:
reasons.append(
"PR head changed after validation; re-pin and re-validate before "
"approval or merge"
)
proven = not reasons
return {
"proven": proven,
"block": not proven,
"reasons": reasons,
"reviewed_head_sha": reviewed,
"live_head_sha": live,
"head_changed": bool(reviewed and live and reviewed != live),
}
def assess_conflict_fix_push(
*,
pr_number: int,
comments: list[dict],
branch_head_before: str | None,
branch_head_after: str | None,
worktree_path: str | None,
push_cwd: str | None,
is_fast_forward: bool | None,
now: datetime | None = None,
) -> dict[str, Any]:
"""Author pre-push gate: block when a reviewer holds an active lease."""
now = now or datetime.now(timezone.utc)
reasons: list[str] = []
reviewer_lease = find_active_reviewer_lease(comments, pr_number=pr_number, now=now)
conflict_lease = find_active_conflict_fix_lease(comments, pr_number=pr_number, now=now)
if reviewer_lease:
reasons.append(
f"active reviewer lease on PR #{pr_number} "
f"(phase={reviewer_lease.get('phase')}); author push blocked"
)
head_before = _normalize_sha(branch_head_before)
head_after = _normalize_sha(branch_head_after)
if not head_before:
reasons.append("branch head before push missing or invalid SHA")
if head_after and head_before and head_before == head_after:
reasons.append("branch head unchanged; no push to perform")
worktree = (worktree_path or "").strip()
cwd = (push_cwd or "").strip()
if not worktree:
reasons.append("worktree path required for conflict-fix push proof")
elif cwd and worktree and not cwd.rstrip("/").endswith(worktree.rstrip("/").split("/")[-1]):
if worktree not in cwd:
reasons.append(
f"push cwd '{cwd}' does not match session worktree '{worktree}'"
)
if is_fast_forward is False:
reasons.append("non-fast-forward push rejected for conflict-fix (fail closed)")
if conflict_lease and conflict_lease.get("phase") == "pushing":
owner = conflict_lease.get("worktree")
if owner and worktree and owner != worktree:
reasons.append(
f"sibling conflict-fix lease active from worktree '{owner}'"
)
push_allowed = not reasons
return {
"push_allowed": push_allowed,
"block": not push_allowed,
"reasons": reasons,
"active_reviewer_lease": reviewer_lease,
"active_conflict_fix_lease": conflict_lease,
"branch_head_before": head_before,
"branch_head_after": head_after,
"reviewer_was_active": bool(reviewer_lease),
"fast_forward": is_fast_forward,
}
def assess_reviewer_mutation_blocked(
*,
pr_number: int,
comments: list[dict],
reviewed_head_sha: str | None,
live_head_sha: str | None,
mutation: str,
now: datetime | None = None,
) -> dict[str, Any]:
"""Reviewer gate: block when conflict-fix lease active or head moved."""
now = now or datetime.now(timezone.utc)
reasons: list[str] = []
conflict_lease = find_active_conflict_fix_lease(comments, pr_number=pr_number, now=now)
if conflict_lease and (conflict_lease.get("phase") or "") in _ACTIVE_CONFLICT_FIX_PHASES:
reasons.append(
f"active conflict-fix lease on PR #{pr_number} "
f"(phase={conflict_lease.get('phase')}); reviewer {mutation} blocked"
)
head_check = assess_head_sha_equality(reviewed_head_sha, live_head_sha)
if head_check["block"]:
reasons.extend(head_check["reasons"])
if not _normalize_sha(reviewed_head_sha):
reasons.append(
f"reviewed head SHA required before reviewer {mutation} (fail closed)"
)
allowed = not reasons
return {
"mutation_allowed": allowed,
"block": not allowed,
"reasons": reasons,
"active_conflict_fix_lease": conflict_lease,
"head_check": head_check,
"reviewed_head_sha": head_check.get("reviewed_head_sha"),
"live_head_sha": head_check.get("live_head_sha"),
"push_during_validation": bool(
conflict_lease and conflict_lease.get("phase") in {"pushing", "pushed"}
),
}
_REVIEWED_HEAD_RE = re.compile(
r"reviewed head sha\s*:\s*([0-9a-f]{40})",
re.IGNORECASE,
)
_LIVE_HEAD_BEFORE_APPROVAL_RE = re.compile(
r"(?:live head sha before approval|final live head sha before approval)\s*:\s*([0-9a-f]{40})",
re.IGNORECASE,
)
_LIVE_HEAD_BEFORE_MERGE_RE = re.compile(
r"(?:live head sha before merge|final live head sha before merge)\s*:\s*([0-9a-f]{40})",
re.IGNORECASE,
)
_PUSH_DURING_VALIDATION_RE = re.compile(
r"push(?:es)? occurred during validation\s*:\s*(yes|no|true|false)",
re.IGNORECASE,
)
_CONFLICT_HEAD_BEFORE_RE = re.compile(
r"branch head before push\s*:\s*([0-9a-f]{40})",
re.IGNORECASE,
)
_CONFLICT_HEAD_AFTER_RE = re.compile(
r"branch head after push\s*:\s*([0-9a-f]{40})",
re.IGNORECASE,
)
_REVIEWER_LEASE_STATUS_RE = re.compile(
r"active reviewer lease status\s*:\s*(.+)$",
re.IGNORECASE | re.MULTILINE,
)
_FAST_FORWARD_RE = re.compile(
r"whether push was fast-forward\s*:\s*(yes|no|true|false)",
re.IGNORECASE,
)
_REVIEWER_ACTIVE_RE = re.compile(
r"whether any reviewer was active\s*:\s*(yes|no|true|false)",
re.IGNORECASE,
)
def assess_reviewer_stale_head_final_report(report_text: str) -> dict[str, Any]:
"""Final-report proof for reviewed vs live head SHAs (#399 AC 6)."""
text = report_text or ""
reasons: list[str] = []
reviewed = _normalize_sha(_REVIEWED_HEAD_RE.search(text).group(1) if _REVIEWED_HEAD_RE.search(text) else None)
live_approval = _normalize_sha(
_LIVE_HEAD_BEFORE_APPROVAL_RE.search(text).group(1)
if _LIVE_HEAD_BEFORE_APPROVAL_RE.search(text)
else None
)
live_merge = _normalize_sha(
_LIVE_HEAD_BEFORE_MERGE_RE.search(text).group(1)
if _LIVE_HEAD_BEFORE_MERGE_RE.search(text)
else None
)
push_during = _PUSH_DURING_VALIDATION_RE.search(text)
if not reviewed:
reasons.append("reviewed head SHA not stated in final report")
if not live_approval:
reasons.append("final live head SHA before approval not stated")
if not live_merge:
reasons.append("final live head SHA before merge not stated")
if not push_during:
reasons.append("whether push occurred during validation not stated")
elif reviewed and live_approval and reviewed != live_approval:
reasons.append("live head before approval differs from reviewed head SHA")
elif reviewed and live_merge and reviewed != live_merge:
reasons.append("live head before merge differs from reviewed head SHA")
proven = not reasons
return {
"proven": proven,
"block": not proven,
"reasons": reasons,
"reviewed_head_sha": reviewed,
"live_head_sha_before_approval": live_approval,
"live_head_sha_before_merge": live_merge,
"push_during_validation": (push_during.group(1).lower() if push_during else None),
}
def assess_conflict_fix_final_report(report_text: str) -> dict[str, Any]:
"""Final-report proof for conflict-fix push sessions (#399 AC 7)."""
text = report_text or ""
reasons: list[str] = []
head_before = _normalize_sha(
_CONFLICT_HEAD_BEFORE_RE.search(text).group(1)
if _CONFLICT_HEAD_BEFORE_RE.search(text)
else None
)
head_after = _normalize_sha(
_CONFLICT_HEAD_AFTER_RE.search(text).group(1)
if _CONFLICT_HEAD_AFTER_RE.search(text)
else None
)
if not head_before:
reasons.append("branch head before push not stated")
if not head_after:
reasons.append("branch head after push not stated")
if not _REVIEWER_LEASE_STATUS_RE.search(text):
reasons.append("active reviewer lease status not stated")
if not _FAST_FORWARD_RE.search(text):
reasons.append("whether push was fast-forward not stated")
if not _REVIEWER_ACTIVE_RE.search(text):
reasons.append("whether any reviewer was active not stated")
proven = not reasons
return {
"proven": proven,
"block": not proven,
"reasons": reasons,
"branch_head_before": head_before,
"branch_head_after": head_after,
}
skills/llm-project-workflow/workflows/review-merge-pr.md
+18
View File
@@ -752,6 +752,24 @@ inventory before continuing.
Final reports must include lease session id, acquisition proof, heartbeat
status, and release/blocked status.
## 26C. Conflict-fix lease and stale-head protection (#399)
Before validating, approving, or merging a PR:
1. Check for an active conflict-fix lease on the PR; stop if one is active.
2. Pin `expected_head_sha` before validation and pass it to
`gitea_mark_final_review_decision`, `gitea_submit_pr_review`, and
`gitea_merge_pr`.
3. Re-fetch live PR head immediately before approval and merge; refuse when
live head differs from the reviewed SHA.
Final reports must state:
* reviewed head SHA
* final live head SHA before approval
* final live head SHA before merge
* whether any push occurred during validation
## 27. Merge rules
Before merge, rerun fresh live checks:
skills/llm-project-workflow/workflows/work-issue.md
+23
View File
@@ -607,6 +607,29 @@ After push, report:
If push fails, stop and produce a recovery handoff.
## 20A. Conflict-fix lease and push gate (#399)
When pushing to an existing PR branch to resolve merge conflicts:
1. Call `gitea_acquire_conflict_fix_lease` before any push.
2. Call `gitea_assess_conflict_fix_push` immediately before `git push` with:
* branch head before push
* branch head after push (local)
* session worktree path
* push cwd
* whether the push is fast-forward
3. Do not push when a reviewer holds an active lease on the same PR.
4. Do not force-push.
5. Do not push from the main checkout or wrong cwd.
Conflict-fix final reports must state:
* branch head before push
* branch head after push
* active reviewer lease status
* whether push was fast-forward
* whether any reviewer was active
## 21. PR creation rules
Create a PR only if implementation and validation pass, unless project policy explicitly allows draft PRs with documented validation failures.
tests/test_audit.py
+28 -16
View File
@@ -284,21 +284,38 @@ class TestSimpleToolAudit(_AuditWiringBase):
self.assertEqual(result["number"], 9)
_NO_PR_WORK_LEASE_BLOCK = {"block": False, "reasons": [], "mutation_allowed": True}
class TestGatedToolAudit(_AuditWiringBase):
def setUp(self):
super().setUp()
from mcp_server import init_review_decision_lock
from tests.test_mcp_server import _install_owned_reviewer_lease
import reviewer_pr_lease
# init_review_decision_lock clears any prior session lease (#407).
init_review_decision_lock("prgs", "review_pr")
self._lease_patch = _install_owned_reviewer_lease(8)
self._lease_patch.start()
self._auth_identity_patch = patch(
"mcp_server._authenticated_username", return_value="reviewer-bot"
)
self._auth_identity_patch.start()
self._pr_lease_comments_patch = patch(
"mcp_server._list_pr_lease_comments", return_value=[]
)
self._pr_lease_comments_patch.start()
self._pr_work_lease_patch = patch(
"mcp_server._pr_work_lease_reviewer_block",
return_value=dict(_NO_PR_WORK_LEASE_BLOCK),
)
self._pr_work_lease_patch.start()
self.addCleanup(self._auth_identity_patch.stop)
self.addCleanup(self._lease_patch.stop)
self.addCleanup(self._pr_lease_comments_patch.stop)
self.addCleanup(self._pr_work_lease_patch.stop)
self.addCleanup(reviewer_pr_lease.clear_session_lease)
def _pr(self, author, state="open", sha="abc123", mergeable=True):
@@ -350,7 +367,9 @@ class TestGatedToolAudit(_AuditWiringBase):
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_submit_review_success_audited(self, _auth, mock_api):
# mark_final_review_decision and submit each run eligibility (user + PR).
mock_api.side_effect = [
{"login": "reviewer-bot"}, self._pr("author-bot"),
{"login": "reviewer-bot"}, self._pr("author-bot"),
{"id": 7, "state": "APPROVED"},
[{"id": 7, "user": {"login": "reviewer-bot"}, "state": "APPROVED",
@@ -359,23 +378,16 @@ class TestGatedToolAudit(_AuditWiringBase):
env = self._env(GITEA_PROFILE_NAME="gitea-reviewer",
GITEA_ALLOWED_OPERATIONS="read,review,approve")
with patch.dict(os.environ, env, clear=True):
from mcp_server import init_review_decision_lock, gitea_mark_final_review_decision
from tests.test_mcp_server import _install_owned_reviewer_lease
import reviewer_pr_lease
from mcp_server import gitea_mark_final_review_decision
init_review_decision_lock("prgs", "review_pr")
gitea_mark_final_review_decision(8, "approve", remote="prgs")
lease_patch = _install_owned_reviewer_lease(8)
lease_patch.start()
try:
r = gitea_submit_pr_review(
pr_number=8, action="approve",
body="LGTM", remote="prgs",
final_review_decision_ready=True,
)
finally:
lease_patch.stop()
reviewer_pr_lease.clear_session_lease()
gitea_mark_final_review_decision(
8, "approve", expected_head_sha="abc123", remote="prgs",
)
r = gitea_submit_pr_review(
pr_number=8, action="approve",
body="LGTM", remote="prgs",
final_review_decision_ready=True,
)
self.assertTrue(r["performed"])
recs = self._records()
self.assertEqual(len(recs), 1)
tests/test_final_report_validator.py
+4
View File
@@ -41,6 +41,10 @@ def _review_handoff(**overrides):
"- Selected PR: #203",
"- Reviewer eligibility: eligible",
"- Pinned reviewed head: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
"- Reviewed head SHA: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
"- Final live head SHA before approval: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
"- Final live head SHA before merge: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
"- Push occurred during validation: no",
"- Worktree path: branches/review-203",
"- Worktree dirty: clean",
"- Scratch worktree used: yes (branches/review-203)",
tests/test_llm_agent_sha.py
+8 -4
View File
@@ -122,15 +122,19 @@ class TestShaCannotBypassSelfReview(unittest.TestCase):
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_review_tool_refuses_self_approval_despite_sha(self, _auth, mock_api, mock_get_all):
mock_get_all.return_value = [{"number": 9, "title": "PR 9", "state": "open", "head": {"ref": "branch9", "sha": "abc1234"}, "base": {"ref": "master"}, "mergeable": True, "user": {"login": "jcwalker3"}}]
from mcp_server import init_review_decision_lock
from tests.test_mcp_server import FULL_HEAD_SHA, _seed_ready_review_decision
head_sha = FULL_HEAD_SHA
mock_get_all.return_value = [{"number": 9, "title": "PR 9", "state": "open", "head": {"ref": "branch9", "sha": head_sha}, "base": {"ref": "master"}, "mergeable": True, "user": {"login": "jcwalker3"}}]
mock_api.side_effect = [
{"login": "jcwalker3"}, # /user (inventory)
{"login": "jcwalker3"}, # /user (submit eligibility)
{"user": {"login": "jcwalker3"}, "state": "open", "head": {"sha": "abc1234"}, "mergeable": True}, # /pulls/9
{"user": {"login": "jcwalker3"}, "state": "open", "head": {"sha": head_sha}, "mergeable": True}, # /pulls/9
]
from mcp_server import init_review_decision_lock, gitea_mark_final_review_decision
init_review_decision_lock("prgs", "review_pr")
gitea_mark_final_review_decision(9, "approve", remote="prgs")
with patch("mcp_server._list_pr_lease_comments", return_value=[]):
_seed_ready_review_decision(9, "approve", sha=head_sha, remote="prgs")
env = self._env(SHA_WOULD_BE_REVIEWER, "reviewer")
with patch.dict(os.environ, env, clear=True):
r = gitea_review_pr(
tests/test_mcp_server.py
+163 -51
View File
@@ -49,6 +49,7 @@ import mcp_server
import issue_lock_store
FAKE_AUTH = "Basic dGVzdDp0ZXN0"
FULL_HEAD_SHA = "a" * 40
_NO_BLOCKER_FEEDBACK = {
"success": True,
@@ -62,6 +63,7 @@ def _mark_request_changes_ready(pr_number=8, **kwargs):
suppression feedback fetch stubbed to 'no existing blocker'."""
with patch("mcp_server.gitea_get_pr_review_feedback",
return_value=dict(_NO_BLOCKER_FEEDBACK)):
kwargs.setdefault("expected_head_sha", "abc123")
return gitea_mark_final_review_decision(
pr_number, "request_changes", **kwargs)
@@ -82,6 +84,7 @@ def _visible_approval_reviews(reviewer="reviewer-bot", sha="abc123"):
_DEFAULT_LEASE_SESSION = "mcp-test-reviewer-lease"
_NO_PR_WORK_LEASE_BLOCK = {"block": False, "reasons": [], "mutation_allowed": True}
def _reviewer_lease_comment(
@@ -139,6 +142,45 @@ def _install_owned_reviewer_lease(
)
def _seed_ready_review_decision(
pr_number,
action,
*,
sha=FULL_HEAD_SHA,
remote="prgs",
org=None,
repo=None,
):
"""Mark the review-decision lock ready without mark_final API calls (#399)."""
import mcp_server as _m
resolved_org, resolved_repo = org, repo
if remote in _m.REMOTES:
_, resolved_org, resolved_repo = _m._resolve(remote, None, org, repo)
profile_name = (_m.get_profile().get("profile_name") or "").strip()
session_lock = (
(os.environ.get(_m.SESSION_PROFILE_LOCK_ENV) or "").strip()
or profile_name
)
_m._save_review_decision_lock({
"task": "review_pr",
"remote": remote,
"session_pid": os.getpid(),
"session_profile": profile_name,
"session_profile_lock": session_lock,
"final_review_decision_ready": True,
"ready_pr_number": pr_number,
"ready_action": action,
"ready_expected_head_sha": sha,
"ready_remote": remote,
"ready_org": resolved_org,
"ready_repo": resolved_repo,
"live_mutations": [],
"correction_authorized": False,
"correction_reason": None,
})
# Issue-write tools are profile-gated (#69).
ISSUE_WRITE_ENV = {
"GITEA_ALLOWED_OPERATIONS": (
@@ -638,6 +680,17 @@ class TestMergePR(unittest.TestCase):
self.addCleanup(self._auth_identity_patch.stop)
self.addCleanup(self._lease_patch.stop)
self.addCleanup(reviewer_pr_lease.clear_session_lease)
self._pr_lease_comments_patch = patch(
"mcp_server._list_pr_lease_comments", return_value=[]
)
self._pr_lease_comments_patch.start()
self.addCleanup(self._pr_lease_comments_patch.stop)
self._pr_work_lease_patch = patch(
"mcp_server._pr_work_lease_reviewer_block",
return_value=dict(_NO_PR_WORK_LEASE_BLOCK),
)
self._pr_work_lease_patch.start()
self.addCleanup(self._pr_work_lease_patch.stop)
def _pr(self, author, state="open", sha="abc123", mergeable=True):
return {
@@ -711,7 +764,8 @@ class TestMergePR(unittest.TestCase):
with patch.dict(os.environ, env, clear=True):
r = gitea_merge_pr(
pr_number=8, confirmation=self._confirm(8),
expected_changed_files=["b.py", "a.py"], remote="prgs")
expected_changed_files=["b.py", "a.py"],
expected_head_sha="abc123", remote="prgs")
self.assertTrue(r["performed"])
# -- read-back / cleanup surfacing (#98) -----------------------------------
@@ -729,8 +783,10 @@ class TestMergePR(unittest.TestCase):
env = {"GITEA_PROFILE_NAME": "gitea-merger",
"GITEA_ALLOWED_OPERATIONS": "read,merge"}
with patch.dict(os.environ, env, clear=True):
r = gitea_merge_pr(pr_number=8, confirmation=self._confirm(8),
remote="prgs")
r = gitea_merge_pr(
pr_number=8, confirmation=self._confirm(8),
expected_head_sha="abc123", remote="prgs",
)
# The merge itself is still reported performed/successful.
self.assertTrue(r["performed"])
self.assertEqual(r["merge_result"], "PR #8 merged via 'merge'.")
@@ -758,8 +814,10 @@ class TestMergePR(unittest.TestCase):
env = {"GITEA_PROFILE_NAME": "gitea-merger",
"GITEA_ALLOWED_OPERATIONS": "read,merge"}
with patch.dict(os.environ, env, clear=True):
r = gitea_merge_pr(pr_number=8, confirmation=self._confirm(8),
remote="prgs")
r = gitea_merge_pr(
pr_number=8, confirmation=self._confirm(8),
expected_head_sha="abc123", remote="prgs",
)
self.assertTrue(r["performed"])
self.assertEqual(r["merge_commit"], "c9")
self.assertTrue(r["cleanup_status"].startswith("skipped (cleanup error:"))
@@ -773,7 +831,7 @@ class TestMergePR(unittest.TestCase):
env = {"GITEA_PROFILE_NAME": "gitea-merger",
"GITEA_ALLOWED_OPERATIONS": "read,merge"}
with patch.dict(os.environ, env, clear=True):
r = gitea_merge_pr(pr_number=8, confirmation="", remote="prgs")
r = gitea_merge_pr(pr_number=8, confirmation="", expected_head_sha="abc123", remote="prgs")
self.assertFalse(r["performed"])
self.assertTrue(any("explicit confirmation required" in x for x in r["reasons"]))
mock_api.assert_not_called()
@@ -784,7 +842,7 @@ class TestMergePR(unittest.TestCase):
env = {"GITEA_PROFILE_NAME": "gitea-merger",
"GITEA_ALLOWED_OPERATIONS": "read,merge"}
with patch.dict(os.environ, env, clear=True):
r = gitea_merge_pr(pr_number=8, confirmation="MERGE PR 9", remote="prgs")
r = gitea_merge_pr(pr_number=8, confirmation="MERGE PR 9", expected_head_sha="abc123", remote="prgs")
self.assertFalse(r["performed"])
mock_api.assert_not_called()
@@ -918,7 +976,8 @@ class TestMergePR(unittest.TestCase):
with patch.dict(os.environ, env, clear=True):
r = gitea_merge_pr(
pr_number=8, confirmation=self._confirm(8),
expected_changed_files=["a.py", "b.py"], remote="prgs")
expected_changed_files=["a.py", "b.py"],
expected_head_sha="abc123", remote="prgs")
self.assertFalse(r["performed"])
self.assertIn(
"PR changed files do not match expected_changed_files (fail closed)",
@@ -931,7 +990,7 @@ class TestMergePR(unittest.TestCase):
def test_invalid_merge_method_rejected(self, mock_api):
with patch.dict(os.environ, {}, clear=True):
r = gitea_merge_pr(
pr_number=8, confirmation="MERGE PR 8", do="octopus", remote="prgs")
pr_number=8, confirmation="MERGE PR 8", do="octopus", expected_head_sha="abc123", remote="prgs")
self.assertFalse(r["performed"])
self.assertTrue(any("unknown merge method" in x for x in r["reasons"]))
mock_api.assert_not_called()
@@ -949,7 +1008,9 @@ class TestMergePR(unittest.TestCase):
"GITEA_TOKEN": "super-secret-token"}
with patch.dict(os.environ, env, clear=True):
r = gitea_merge_pr(
pr_number=8, confirmation=self._confirm(8), remote="prgs")
pr_number=8, confirmation=self._confirm(8),
expected_head_sha="abc123", remote="prgs",
)
blob = repr(r).lower()
for secret in ("super-secret-token", "authorization", "basic ", FAKE_AUTH.lower()):
self.assertNotIn(secret, blob)
@@ -966,7 +1027,9 @@ class TestMergePR(unittest.TestCase):
"GITEA_ALLOWED_OPERATIONS": "read,merge"}
with patch.dict(os.environ, env, clear=True):
r = gitea_merge_pr(
pr_number=8, confirmation=self._confirm(8), remote="prgs")
pr_number=8, confirmation=self._confirm(8),
expected_head_sha="abc123", remote="prgs",
)
self.assertFalse(r["performed"])
blob = repr(r)
self.assertIn("[REDACTED]", blob)
@@ -1007,7 +1070,9 @@ class TestMergePR(unittest.TestCase):
"GITEA_ALLOWED_OPERATIONS": "read,merge"}
with patch.dict(os.environ, env, clear=True):
r = gitea_merge_pr(
pr_number=8, confirmation=self._confirm(8), remote="prgs")
pr_number=8, confirmation=self._confirm(8),
expected_head_sha="abc123", remote="prgs",
)
self.assertFalse(r["performed"])
self.assertFalse(r.get("approval_visible"))
self.assertTrue(any("no visible APPROVED review" in x for x in r["reasons"]))
@@ -1028,7 +1093,9 @@ class TestMergePR(unittest.TestCase):
"GITEA_ALLOWED_OPERATIONS": "read,merge"}
with patch.dict(os.environ, env, clear=True):
r = gitea_merge_pr(
pr_number=8, confirmation=self._confirm(8), remote="prgs")
pr_number=8, confirmation=self._confirm(8),
expected_head_sha="abc123", remote="prgs",
)
self.assertFalse(r["performed"])
self.assertTrue(r.get("has_blocking_change_requests"))
self.assertTrue(any("REQUEST_CHANGES" in x for x in r["reasons"]))
@@ -1129,16 +1196,18 @@ class TestReviewPR(unittest.TestCase):
"forbidden_operations": [],
"base_url": None,
}
mock_get_all.return_value = [{"number": 1, "title": "PR 1", "state": "open", "head": {"ref": "branch1", "sha": "abc1234"}, "base": {"ref": "master"}, "mergeable": True, "user": {"login": "jcwalker3"}}]
head_sha = FULL_HEAD_SHA
mock_get_all.return_value = [{"number": 1, "title": "PR 1", "state": "open", "head": {"ref": "branch1", "sha": head_sha}, "base": {"ref": "master"}, "mergeable": True, "user": {"login": "jcwalker3"}}]
# mock_api responses: 1) /user (inventory), 2) /user (eligibility), 3) /pulls/1 (eligibility)
mock_api.side_effect = [
{"login": "jcwalker3"}, # /api/v1/user (inventory)
{"login": "jcwalker3"}, # /api/v1/user (submit eligibility)
{"user": {"login": "jcwalker3"}, "state": "open", "head": {"sha": "abc1234"}, "mergeable": True}, # /pulls/1
{"user": {"login": "jcwalker3"}, "state": "open", "head": {"sha": head_sha}, "mergeable": True}, # /pulls/1
]
from mcp_server import init_review_decision_lock
init_review_decision_lock("prgs", "review_pr")
gitea_mark_final_review_decision(1, "approve", remote="prgs")
with patch("mcp_server._list_pr_lease_comments", return_value=[]):
_seed_ready_review_decision(1, "approve", sha=head_sha, remote="prgs")
result = gitea_review_pr(
pr_number=1,
event="APPROVE",
@@ -1794,7 +1863,7 @@ class TestReviewDecisionValidationGate(unittest.TestCase):
"""Block incidental live review mutations during validation."""
PR = 203
SHA = "abc123"
SHA = FULL_HEAD_SHA
def _pr(self, author, sha=SHA):
return {
@@ -1819,6 +1888,18 @@ class TestReviewDecisionValidationGate(unittest.TestCase):
self.addCleanup(self._auth_identity_patch.stop)
self.addCleanup(self._lease_patch.stop)
self.addCleanup(reviewer_pr_lease.clear_session_lease)
self._pr_lease_comments_patch = patch(
"mcp_server._list_pr_lease_comments", return_value=[]
)
self._pr_lease_comments_patch.start()
self._pr_work_lease_patch = patch(
"mcp_server._pr_work_lease_reviewer_block",
return_value=dict(_NO_PR_WORK_LEASE_BLOCK),
)
self._pr_work_lease_patch.start()
self.addCleanup(self._pr_lease_comments_patch.stop)
self.addCleanup(self._pr_work_lease_patch.stop)
def _env(self):
return patch.dict(os.environ, {
@@ -1916,15 +1997,28 @@ class TestSubmitPrReview(unittest.TestCase):
import reviewer_pr_lease
init_review_decision_lock("prgs", "review_pr")
gitea_mark_final_review_decision(8, "approve", remote="prgs")
self._lease_patch = _install_owned_reviewer_lease(8)
self._lease_patch.start()
self._auth_identity_patch = patch(
"mcp_server._authenticated_username", return_value="reviewer-bot"
)
self._auth_identity_patch.start()
self._pr_lease_comments_patch = patch(
"mcp_server._list_pr_lease_comments", return_value=[]
)
self._pr_lease_comments_patch.start()
self._pr_work_lease_patch = patch(
"mcp_server._pr_work_lease_reviewer_block",
return_value=dict(_NO_PR_WORK_LEASE_BLOCK),
)
self._pr_work_lease_patch.start()
gitea_mark_final_review_decision(
8, "approve", remote="prgs", expected_head_sha="abc123",
)
self.addCleanup(self._auth_identity_patch.stop)
self.addCleanup(self._lease_patch.stop)
self.addCleanup(self._pr_lease_comments_patch.stop)
self.addCleanup(self._pr_work_lease_patch.stop)
self.addCleanup(reviewer_pr_lease.clear_session_lease)
def _pr(self, author, state="open", sha="abc123", mergeable=True):
@@ -2085,10 +2179,11 @@ class TestSubmitPrReview(unittest.TestCase):
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_comment_succeeds_when_review_eligible(self, _auth, mock_api):
gitea_mark_final_review_decision(8, "comment", remote="prgs")
mock_api.side_effect = [
{"login": "reviewer-bot"}, self._pr("author-bot"),
{"login": "reviewer-bot"}, self._pr("author-bot"), {"id": 3},
]
gitea_mark_final_review_decision(8, "comment", remote="prgs", expected_head_sha="abc123")
env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
"GITEA_ALLOWED_OPERATIONS": "read,review"}
with patch.dict(os.environ, env, clear=True):
@@ -2104,12 +2199,15 @@ class TestSubmitPrReview(unittest.TestCase):
with patch("mcp_server.get_auth_header", return_value=FAKE_AUTH), \
patch("mcp_server.api_request") as mock_api:
mock_api.side_effect = [
{"login": "jcwalker3"}, self._pr("jcwalker3"),
{"login": "jcwalker3"}, self._pr("jcwalker3"), {"id": 4},
]
env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
"GITEA_ALLOWED_OPERATIONS": "read,review"}
with patch.dict(os.environ, env, clear=True):
gitea_mark_final_review_decision(8, "comment", remote="prgs")
gitea_mark_final_review_decision(
8, "comment", remote="prgs", expected_head_sha="abc123",
)
r = gitea_submit_pr_review(
pr_number=8, action="comment", body="note", remote="prgs",
final_review_decision_ready=True,
@@ -2152,14 +2250,14 @@ class TestSubmitPrReview(unittest.TestCase):
with patch("mcp_server.get_auth_header", return_value=FAKE_AUTH), \
patch("mcp_server.api_request") as mock_api:
mock_api.side_effect = [
{"login": "reviewer-bot"}, self._pr("author-bot", sha="abc123"),
{"login": "reviewer-bot"}, self._pr("author-bot", sha="deadbeef"),
]
env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
"GITEA_ALLOWED_OPERATIONS": "read,review,approve"}
with patch.dict(os.environ, env, clear=True):
r = gitea_submit_pr_review(
pr_number=8, action="approve",
expected_head_sha="deadbeef", remote="prgs",
expected_head_sha="abc123", remote="prgs",
final_review_decision_ready=True,
)
self.assertFalse(r["performed"])
@@ -2210,7 +2308,7 @@ class TestSubmitPrReview(unittest.TestCase):
"GITEA_ALLOWED_OPERATIONS": "read,review,approve",
"GITEA_TOKEN": "super-secret-token"}
with patch.dict(os.environ, env, clear=True):
gitea_mark_final_review_decision(5, "approve", remote="prgs")
gitea_mark_final_review_decision(5, "approve", remote="prgs", expected_head_sha="abc123")
r = gitea_submit_pr_review(
pr_number=5, action="approve", remote="prgs",
final_review_decision_ready=True,
@@ -2230,7 +2328,6 @@ class TestSubmitPrReview(unittest.TestCase):
env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
"GITEA_ALLOWED_OPERATIONS": "read,review,approve"}
with patch.dict(os.environ, env, clear=True):
gitea_mark_final_review_decision(8, "approve", remote="prgs")
r = gitea_submit_pr_review(
pr_number=8, action="approve", remote="prgs",
final_review_decision_ready=True,
@@ -2320,7 +2417,7 @@ class TestSubmitPrReview(unittest.TestCase):
def test_mark_final_decision_rejects_remote_mismatch(self):
init_review_decision_lock("prgs", "review_pr")
r = gitea_mark_final_review_decision(8, "approve", remote="dadeschools")
r = gitea_mark_final_review_decision(8, "approve", remote="dadeschools", expected_head_sha="abc123")
self.assertFalse(r["marked_ready"])
self.assertTrue(any("does not match locked remote" in x for x in r["reasons"]))
@@ -2332,28 +2429,30 @@ class TestSubmitPrReview(unittest.TestCase):
{"id": 42, "state": "APPROVED"},
[_formal_review("reviewer-bot", "APPROVED", review_id=42)],
{"login": "reviewer-bot"}, self._pr("author-bot"),
{"login": "reviewer-bot"}, self._pr("author-bot"),
{"id": 43, "state": "REQUEST_CHANGES"},
[_formal_review("reviewer-bot", "REQUEST_CHANGES", review_id=43)],
]
env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
"GITEA_ALLOWED_OPERATIONS": "read,review,approve,request_changes"}
gitea_mark_final_review_decision(8, "approve", remote="prgs")
with patch.dict(os.environ, env, clear=True):
first = gitea_submit_pr_review(
pr_number=8, action="approve", remote="prgs",
final_review_decision_ready=True,
)
auth = gitea_authorize_review_correction(
prior_review_id=42,
prior_review_state="approve",
reason="operator approved correcting mistaken approve",
operator_authorized=True,
)
_mark_request_changes_ready(remote="prgs")
second = gitea_submit_pr_review(
pr_number=8, action="request_changes", remote="prgs",
final_review_decision_ready=True,
)
with patch("mcp_server.gitea_get_pr_review_feedback",
return_value=dict(_NO_BLOCKER_FEEDBACK)):
with patch.dict(os.environ, env, clear=True):
first = gitea_submit_pr_review(
pr_number=8, action="approve", remote="prgs",
final_review_decision_ready=True,
)
auth = gitea_authorize_review_correction(
prior_review_id=42,
prior_review_state="approve",
reason="operator approved correcting mistaken approve",
operator_authorized=True,
)
_mark_request_changes_ready(remote="prgs")
second = gitea_submit_pr_review(
pr_number=8, action="request_changes", remote="prgs",
final_review_decision_ready=True,
)
self.assertTrue(first["performed"])
self.assertTrue(auth["authorized"])
self.assertTrue(second["performed"])
@@ -2365,7 +2464,7 @@ class TestSubmitPrReview(unittest.TestCase):
"GITEA_ALLOWED_OPERATIONS": "read,review,approve"}
with patch.dict(os.environ, env, clear=True):
# Mark decision for specific remote, org, repo
gitea_mark_final_review_decision(8, "approve", remote="prgs", org="MyOrg", repo="MyRepo")
gitea_mark_final_review_decision(8, "approve", remote="prgs", expected_head_sha="abc123", org="MyOrg", repo="MyRepo")
# Mismatched remote
r = gitea_submit_pr_review(
@@ -2407,6 +2506,11 @@ class TestTrackerHygieneCleanup(unittest.TestCase):
self.mock_audit = patch("gitea_audit.write_event").start()
# gitea.pr.close: closing a PR via gitea_edit_pr is capability-gated (#216).
patch("mcp_server.get_profile", return_value={"profile_name": "test", "allowed_operations": ["read", "merge", "edit", "close", "gitea.pr.close", "gitea.issue.close"], "audit_label": "test", "forbidden_operations": []}).start()
patch("mcp_server._list_pr_lease_comments", return_value=[]).start()
patch(
"mcp_server._pr_work_lease_reviewer_block",
return_value=dict(_NO_PR_WORK_LEASE_BLOCK),
).start()
def tearDown(self):
patch.stopall()
@@ -2453,7 +2557,8 @@ class TestTrackerHygieneCleanup(unittest.TestCase):
def test_merge_pr_with_closes_removes_label(self):
import reviewer_pr_lease
lease_patch = _install_owned_reviewer_lease(1, head_sha="sha123")
head_sha = FULL_HEAD_SHA
lease_patch = _install_owned_reviewer_lease(1, head_sha=head_sha)
lease_patch.start()
self.addCleanup(lease_patch.stop)
self.addCleanup(reviewer_pr_lease.clear_session_lease)
@@ -2462,12 +2567,12 @@ class TestTrackerHygieneCleanup(unittest.TestCase):
if method == "GET" and "/user" in url:
return {"login": "merger"}
if method == "GET" and url.endswith("/reviews"):
return [_formal_review("reviewer", "APPROVED", sha="sha123")]
return [_formal_review("reviewer", "APPROVED", sha=head_sha)]
if method == "GET" and "pulls/1" in url and "/files" not in url:
return {
"user": {"login": "author"},
"state": "open",
"head": {"sha": "sha123", "ref": "feat/my-branch"},
"head": {"sha": head_sha, "ref": "feat/my-branch"},
"base": {"ref": "main"},
"mergeable": True,
"merged_commit_sha": "merge123",
@@ -2487,14 +2592,18 @@ class TestTrackerHygieneCleanup(unittest.TestCase):
return {}
self.mock_api.side_effect = api_side_effect
res = gitea_merge_pr(pr_number=1, confirmation="MERGE PR 1", do="merge")
res = gitea_merge_pr(
pr_number=1, confirmation="MERGE PR 1", do="merge",
expected_head_sha=head_sha,
)
self.assertTrue(res["performed"])
self.assertEqual(res["cleanup_status"].get(123), "released")
def test_merge_pr_with_branch_name_removes_label(self):
import reviewer_pr_lease
lease_patch = _install_owned_reviewer_lease(1, head_sha="sha123")
head_sha = FULL_HEAD_SHA
lease_patch = _install_owned_reviewer_lease(1, head_sha=head_sha)
lease_patch.start()
self.addCleanup(lease_patch.stop)
self.addCleanup(reviewer_pr_lease.clear_session_lease)
@@ -2503,12 +2612,12 @@ class TestTrackerHygieneCleanup(unittest.TestCase):
if method == "GET" and "/user" in url:
return {"login": "merger"}
if method == "GET" and url.endswith("/reviews"):
return [_formal_review("reviewer", "APPROVED", sha="sha123")]
return [_formal_review("reviewer", "APPROVED", sha=head_sha)]
if method == "GET" and "pulls/1" in url and "/files" not in url:
return {
"user": {"login": "author"},
"state": "open",
"head": {"sha": "sha123", "ref": "fix/issue-123-slug"},
"head": {"sha": head_sha, "ref": "fix/issue-123-slug"},
"base": {"ref": "main"},
"mergeable": True,
"merged_commit_sha": "merge123",
@@ -2528,7 +2637,10 @@ class TestTrackerHygieneCleanup(unittest.TestCase):
return {}
self.mock_api.side_effect = api_side_effect
res = gitea_merge_pr(pr_number=1, confirmation="MERGE PR 1", do="merge")
res = gitea_merge_pr(
pr_number=1, confirmation="MERGE PR 1", do="merge",
expected_head_sha=head_sha,
)
self.assertTrue(res["performed"])
self.assertEqual(res["cleanup_status"].get(123), "released")
tests/test_permission_reports.py
+6 -2
View File
@@ -230,7 +230,9 @@ class TestEligibilityDenialReport(PermissionReportBase):
return PR_PAYLOAD
mock_api.side_effect = fake_api
mcp_server.init_review_decision_lock("prgs", "review_pr")
mcp_server.gitea_mark_final_review_decision(42, "approve", remote="prgs")
from tests.test_mcp_server import _seed_ready_review_decision
_seed_ready_review_decision(42, "approve", remote="prgs")
with patch.dict(os.environ, self._env("author-profile")):
res = mcp_server.gitea_submit_pr_review(
pr_number=42, action="approve", body="lgtm", remote="prgs",
@@ -272,7 +274,9 @@ class TestReviewCommentPathUsesCanonicalOp(PermissionReportBase):
return PR_PAYLOAD
mock_api.side_effect = fake_api
mcp_server.init_review_decision_lock("prgs", "review_pr")
mcp_server.gitea_mark_final_review_decision(42, "comment", remote="prgs")
from tests.test_mcp_server import _seed_ready_review_decision
_seed_ready_review_decision(42, "comment", remote="prgs")
with patch.dict(os.environ, self._env("author-profile")):
res = mcp_server.gitea_submit_pr_review(
pr_number=42, action="comment", body="finding", remote="prgs",
tests/test_pr_queue_inventory.py
+20 -12
View File
@@ -128,18 +128,29 @@ class TestPRQueueInventory(unittest.TestCase):
"forbidden_operations": [],
"base_url": None,
}
from mcp_server import init_review_decision_lock
from tests.test_mcp_server import (
FULL_HEAD_SHA,
_install_owned_reviewer_lease,
_seed_ready_review_decision,
)
import reviewer_pr_lease
head_sha = FULL_HEAD_SHA
mock_fetch.return_value = _final_page_fetch([
{"number": 1, "title": "PR 1", "state": "open", "head": {"ref": "branch1", "sha": "abc1"}, "base": {"ref": "master"}, "mergeable": True, "user": {"login": "other_user"}}
{"number": 1, "title": "PR 1", "state": "open",
"head": {"ref": "branch1", "sha": head_sha},
"base": {"ref": "master"}, "mergeable": True,
"user": {"login": "other_user"}}
])
# mock_api: inventory whoami, eligibility whoami, eligibility PR,
# POST review (#244: state + visible-verdict GET reviews).
mock_api.side_effect = [
{"login": "reviewer1"},
{"login": "reviewer1"},
{
"user": {"login": "other_user"},
"state": "open",
"head": {"sha": "abc1"},
"head": {"sha": head_sha},
"mergeable": True,
},
{"id": 100, "state": "APPROVED"},
@@ -148,22 +159,19 @@ class TestPRQueueInventory(unittest.TestCase):
"id": 100,
"user": {"login": "reviewer1"},
"state": "APPROVED",
"commit_id": "abc1",
"commit_id": head_sha,
"submitted_at": "2026-07-06T10:00:00Z",
"dismissed": False,
}
],
]
from mcp_server import init_review_decision_lock, gitea_mark_final_review_decision
from tests.test_mcp_server import _install_owned_reviewer_lease
import reviewer_pr_lease
with patch("mcp_server._authenticated_username", return_value="reviewer1"):
with patch("mcp_server._authenticated_username", return_value="reviewer1"), \
patch("mcp_server._list_pr_lease_comments", return_value=[]):
init_review_decision_lock("prgs", "review_pr")
gitea_mark_final_review_decision(1, "approve", remote="prgs")
_seed_ready_review_decision(1, "approve", sha=head_sha, remote="prgs")
lease_patch = _install_owned_reviewer_lease(
1, head_sha="abc1", session_id="inventory-review-lease",
1, head_sha=head_sha, session_id="inventory-review-lease",
)
lease_patch.start()
self.addCleanup(lease_patch.stop)
tests/test_pr_work_lease.py
+207
View File
@@ -0,0 +1,207 @@
#!/usr/bin/env python3
"""Regression tests for conflict-fix and reviewer PR work leases (#399)."""
from __future__ import annotations
import os
import sys
import unittest
from datetime import datetime, timedelta, timezone
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from pr_work_lease import ( # noqa: E402
CONFLICT_FIX_LEASE_MARKER,
REVIEWER_LEASE_MARKER,
assess_conflict_fix_final_report,
assess_conflict_fix_push,
assess_head_sha_equality,
assess_reviewer_mutation_blocked,
assess_reviewer_stale_head_final_report,
format_conflict_fix_lease_body,
parse_conflict_fix_lease_comment,
parse_reviewer_lease_comment,
)
HEAD_A = "a" * 40
HEAD_B = "b" * 40
NOW = datetime(2026, 7, 7, 15, 0, tzinfo=timezone.utc)
def _reviewer_lease_body(*, phase: str = "validating", expires_minutes: int = 60) -> str:
expires = (NOW + timedelta(minutes=expires_minutes)).isoformat().replace("+00:00", "Z")
return "\n".join([
REVIEWER_LEASE_MARKER,
"pr: #376",
"phase: " + phase,
f"candidate_head: {HEAD_A}",
f"expires_at: {expires}",
"profile: prgs-reviewer",
])
def _conflict_fix_body(*, phase: str = "claimed", worktree: str = "branches/fix-376") -> str:
expires = (NOW + timedelta(minutes=60)).isoformat().replace("+00:00", "Z")
return "\n".join([
CONFLICT_FIX_LEASE_MARKER,
"pr: #376",
f"phase: {phase}",
f"worktree: {worktree}",
f"head_before: {HEAD_A}",
f"expires_at: {expires}",
"profile: prgs-author",
])
class TestLeaseParsing(unittest.TestCase):
def test_parse_reviewer_lease(self):
parsed = parse_reviewer_lease_comment(_reviewer_lease_body())
self.assertEqual(parsed["pr_number"], 376)
self.assertEqual(parsed["phase"], "validating")
self.assertEqual(parsed["candidate_head"], HEAD_A)
def test_parse_conflict_fix_lease(self):
parsed = parse_conflict_fix_lease_comment(_conflict_fix_body())
self.assertEqual(parsed["pr_number"], 376)
self.assertEqual(parsed["phase"], "claimed")
class TestConflictFixPushGate(unittest.TestCase):
def test_blocks_push_during_active_reviewer_lease(self):
comments = [{"body": _reviewer_lease_body()}]
result = assess_conflict_fix_push(
pr_number=376,
comments=comments,
branch_head_before=HEAD_A,
branch_head_after=HEAD_B,
worktree_path="branches/fix-376",
push_cwd="/proj/branches/fix-376",
is_fast_forward=True,
now=NOW,
)
self.assertFalse(result["push_allowed"])
self.assertTrue(any("reviewer lease" in r for r in result["reasons"]))
def test_rejects_non_fast_forward(self):
result = assess_conflict_fix_push(
pr_number=376,
comments=[],
branch_head_before=HEAD_A,
branch_head_after=HEAD_B,
worktree_path="branches/fix-376",
push_cwd="/proj/branches/fix-376",
is_fast_forward=False,
now=NOW,
)
self.assertFalse(result["push_allowed"])
self.assertTrue(any("non-fast-forward" in r for r in result["reasons"]))
def test_wrong_cwd_push_attempt(self):
result = assess_conflict_fix_push(
pr_number=376,
comments=[],
branch_head_before=HEAD_A,
branch_head_after=HEAD_B,
worktree_path="branches/fix-376",
push_cwd="/proj/master",
is_fast_forward=True,
now=NOW,
)
self.assertFalse(result["push_allowed"])
self.assertTrue(any("cwd" in r.lower() for r in result["reasons"]))
def test_sibling_conflict_fix_collision(self):
comments = [{"body": _conflict_fix_body(phase="pushing", worktree="branches/other")}]
result = assess_conflict_fix_push(
pr_number=376,
comments=comments,
branch_head_before=HEAD_A,
branch_head_after=HEAD_B,
worktree_path="branches/fix-376",
push_cwd="/proj/branches/fix-376",
is_fast_forward=True,
now=NOW,
)
self.assertFalse(result["push_allowed"])
self.assertTrue(any("sibling conflict-fix" in r for r in result["reasons"]))
class TestReviewerMutationGate(unittest.TestCase):
def test_blocks_review_during_conflict_fix(self):
comments = [{"body": _conflict_fix_body(phase="pushing")}]
result = assess_reviewer_mutation_blocked(
pr_number=376,
comments=comments,
reviewed_head_sha=HEAD_A,
live_head_sha=HEAD_A,
mutation="approve",
now=NOW,
)
self.assertFalse(result["mutation_allowed"])
self.assertTrue(any("conflict-fix lease" in r for r in result["reasons"]))
def test_stale_head_blocks_approval(self):
result = assess_reviewer_mutation_blocked(
pr_number=376,
comments=[],
reviewed_head_sha=HEAD_A,
live_head_sha=HEAD_B,
mutation="merge",
now=NOW,
)
self.assertFalse(result["mutation_allowed"])
self.assertTrue(result["head_check"]["head_changed"])
def test_head_equality_required_fields(self):
result = assess_head_sha_equality(HEAD_A, HEAD_B)
self.assertFalse(result["proven"])
self.assertTrue(result["head_changed"])
class TestFinalReportProof(unittest.TestCase):
def test_reviewer_stale_head_report_requires_fields(self):
result = assess_reviewer_stale_head_final_report("no head proof here")
self.assertFalse(result["proven"])
def test_reviewer_stale_head_report_passes(self):
report = "\n".join([
f"Reviewed head SHA: {HEAD_A}",
f"Final live head SHA before approval: {HEAD_A}",
f"Final live head SHA before merge: {HEAD_A}",
"Push occurred during validation: no",
])
result = assess_reviewer_stale_head_final_report(report)
self.assertTrue(result["proven"])
def test_conflict_fix_report_requires_fields(self):
result = assess_conflict_fix_final_report("incomplete")
self.assertFalse(result["proven"])
def test_conflict_fix_report_passes(self):
report = "\n".join([
f"Branch head before push: {HEAD_A}",
f"Branch head after push: {HEAD_B}",
"Active reviewer lease status: none",
"Whether push was fast-forward: yes",
"Whether any reviewer was active: no",
])
result = assess_conflict_fix_final_report(report)
self.assertTrue(result["proven"])
class TestFormatLease(unittest.TestCase):
def test_format_conflict_fix_lease_includes_marker(self):
body = format_conflict_fix_lease_body(
pr_number=376,
branch="feat/x",
worktree="branches/fix-376",
profile="prgs-author",
head_before=HEAD_A,
)
self.assertIn(CONFLICT_FIX_LEASE_MARKER, body)
parsed = parse_conflict_fix_lease_comment(body)
self.assertEqual(parsed["pr_number"], 376)
if __name__ == "__main__":
unittest.main()
tests/test_review_final_report_schema.py
+6 -2
View File
@@ -19,7 +19,11 @@ def _minimal_review_report(**overrides):
"- Issue/PR: #182 / PR #203",
"- Branch/SHA: feat/x @ 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
"- Files changed: review_proofs.py",
"- Validation: pytest tests/test_review_proofs.py -q in branches/review-203",
"- Validation: pass: pytest tests/test_review_proofs.py -q in branches/review-203",
"- Reviewed head SHA: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
"- Final live head SHA before approval: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
"- Final live head SHA before merge: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
"- Push occurred during validation: no",
"- Mutations: review only",
"- File edits by reviewer: none",
"- Worktree/index mutations: none",
@@ -80,7 +84,7 @@ class TestReviewFinalReportSchema(unittest.TestCase):
def test_reviewed_head_without_validation_blocks(self):
report = _minimal_review_report().replace(
"- Validation: pytest tests/test_review_proofs.py -q in branches/review-203",
"- Validation: pass: pytest tests/test_review_proofs.py -q in branches/review-203",
"- Validation: not run",
)
report += "\n- Pinned reviewed head: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9"
tests/test_terminal_review_hard_stop.py
+18 -11
View File
@@ -12,6 +12,8 @@ from unittest.mock import patch
import mcp_server
HEAD_SHA = "a" * 40
def _lock(mutations=None, correction=False):
return {
@@ -127,10 +129,20 @@ def _feedback(blocking, stale=False, success=True):
"success": success,
"has_blocking_change_requests": blocking,
"review_feedback_stale": stale,
"current_head_sha": "abc123",
"current_head_sha": HEAD_SHA,
}
def _mark(action, pr_number=6, **kwargs):
kwargs.setdefault("expected_head_sha", HEAD_SHA)
no_lease_block = {"block": False, "reasons": [], "mutation_allowed": True}
with patch("mcp_server._list_pr_lease_comments", return_value=[]), \
patch("mcp_server._pr_work_lease_reviewer_block", return_value=no_lease_block):
return mcp_server.gitea_mark_final_review_decision(
pr_number=pr_number, action=action, remote="prgs", **kwargs
)
class TestDuplicateRequestChangesSuppression(unittest.TestCase):
def tearDown(self):
mcp_server._save_review_decision_lock(None)
@@ -139,8 +151,7 @@ class TestDuplicateRequestChangesSuppression(unittest.TestCase):
_seed()
with patch.object(mcp_server, "gitea_get_pr_review_feedback",
return_value=_feedback(blocking=True, stale=False)):
result = mcp_server.gitea_mark_final_review_decision(
pr_number=6, action="request_changes", remote="prgs")
result = _mark("request_changes")
self.assertFalse(result["marked_ready"])
self.assertTrue(
any("duplicate" in r for r in result["reasons"]),
@@ -150,16 +161,14 @@ class TestDuplicateRequestChangesSuppression(unittest.TestCase):
_seed()
with patch.object(mcp_server, "gitea_get_pr_review_feedback",
return_value=_feedback(blocking=True, stale=True)):
result = mcp_server.gitea_mark_final_review_decision(
pr_number=6, action="request_changes", remote="prgs")
result = _mark("request_changes")
self.assertTrue(result["marked_ready"], result.get("reasons"))
def test_request_changes_allowed_when_no_blocker(self):
_seed()
with patch.object(mcp_server, "gitea_get_pr_review_feedback",
return_value=_feedback(blocking=False)):
result = mcp_server.gitea_mark_final_review_decision(
pr_number=6, action="request_changes", remote="prgs")
result = _mark("request_changes")
self.assertTrue(result["marked_ready"], result.get("reasons"))
def test_request_changes_fails_closed_when_feedback_unavailable(self):
@@ -167,8 +176,7 @@ class TestDuplicateRequestChangesSuppression(unittest.TestCase):
with patch.object(mcp_server, "gitea_get_pr_review_feedback",
return_value=_feedback(blocking=False,
success=False)):
result = mcp_server.gitea_mark_final_review_decision(
pr_number=6, action="request_changes", remote="prgs")
result = _mark("request_changes")
self.assertFalse(result["marked_ready"])
self.assertTrue(
any("could not verify" in r for r in result["reasons"]),
@@ -178,8 +186,7 @@ class TestDuplicateRequestChangesSuppression(unittest.TestCase):
_seed()
with patch.object(mcp_server, "gitea_get_pr_review_feedback",
side_effect=AssertionError("must not be called")):
result = mcp_server.gitea_mark_final_review_decision(
pr_number=6, action="approve", remote="prgs")
result = _mark("approve")
self.assertTrue(result["marked_ready"], result.get("reasons"))