feat: add per-PR reviewer leases for parallel review (Closes #407)

Structured PR-thread leases with acquire/heartbeat MCP tools and
mutation-time enforcement so reviewers cannot race the same PR queue.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
This commit is contained in:
2026-07-07 17:16:06 -04:00
co-authored by Claude Opus 4.8
parent cc4741a4ce
commit 06c2069234
4 changed files with 879 additions and 0 deletions
+284
View File
@@ -543,6 +543,7 @@ import already_landed_reconcile # noqa: E402
import author_mutation_worktree # noqa: E402
import issue_claim_heartbeat # noqa: E402
import issue_work_duplicate_gate # noqa: E402
import reviewer_pr_lease # noqa: E402
import merged_cleanup_reconcile # noqa: E402
import reconciler_profile # noqa: E402
import reconciliation_workflow # noqa: E402
@@ -2016,6 +2017,7 @@ def init_review_decision_lock(remote: str | None, task: str | None):
(os.environ.get(SESSION_PROFILE_LOCK_ENV) or "").strip()
or profile_name
)
reviewer_pr_lease.clear_session_lease()
_save_review_decision_lock({
"task": task,
"remote": remote,
@@ -2458,6 +2460,20 @@ def _evaluate_pr_review_submission(
result["permission_report"] = elig["permission_report"]
return result
if live:
reasons.extend(_reviewer_pr_lease_gate(
pr_number=pr_number,
remote=remote,
host=host,
org=org,
repo=repo,
mutation=action,
live_head_sha=result.get("head_sha"),
pinned_head_sha=expected_head_sha,
))
if reasons:
return result
auth_user = result["authenticated_user"]
pr_author = result["pr_author"]
if action == "approve" and auth_user and pr_author and auth_user == pr_author:
@@ -3257,6 +3273,19 @@ def gitea_merge_pr(
result["permission_report"] = elig["permission_report"]
return result
reasons.extend(_reviewer_pr_lease_gate(
pr_number=pr_number,
remote=remote,
host=host,
org=org,
repo=repo,
mutation="merge",
live_head_sha=result.get("head_sha"),
pinned_head_sha=expected_head_sha,
))
if reasons:
return result
# Gate 4 — head SHA must match if the caller pinned a reviewed SHA.
actual_sha = result["head_sha"]
if expected_head_sha and actual_sha and expected_head_sha != actual_sha:
@@ -4556,6 +4585,261 @@ def _namespace_mutation_block(mutation_task: str, **extra_fields) -> dict | None
return blocked
def _fetch_pr_comments(
pr_number: int,
*,
remote: str,
host: str | None,
org: str | None,
repo: str | None,
) -> list[dict]:
h, o, r = _resolve(remote, host, org, repo)
auth = _auth(h)
api = f"{repo_api_url(h, o, r)}/issues/{pr_number}/comments"
return api_request("GET", api, auth) or []
def _reviewer_pr_lease_gate(
*,
pr_number: int,
remote: str,
host: str | None,
org: str | None,
repo: str | None,
mutation: str,
live_head_sha: str | None,
pinned_head_sha: str | None,
) -> list[str]:
"""Return block reasons when the session lacks an owned PR reviewer lease."""
session = reviewer_pr_lease.get_session_lease()
session_id = (session or {}).get("session_id")
identity = _authenticated_username(remote) or ""
try:
comments = _fetch_pr_comments(
pr_number, remote=remote, host=host, org=org, repo=repo)
except Exception as exc:
return [f"cannot fetch PR comments for lease gate: {_redact(str(exc))}"]
assessment = reviewer_pr_lease.assess_mutation_lease_gate(
pr_number=pr_number,
comments=comments,
reviewer_identity=identity,
session_id=session_id,
mutation=mutation,
live_head_sha=live_head_sha,
pinned_head_sha=pinned_head_sha,
)
return list(assessment.get("reasons") or []) if assessment.get("block") else []
@mcp.tool()
def gitea_acquire_reviewer_pr_lease(
pr_number: int,
worktree: str,
candidate_head: str | None = None,
target_branch: str = "master",
target_branch_sha: str | None = None,
issue_number: int | None = None,
session_id: str | None = None,
remote: str = "dadeschools",
host: str | None = None,
org: str | None = None,
repo: str | None = None,
) -> dict:
"""Acquire a per-PR reviewer lease before review/merge mutations (#407)."""
read_block = _profile_operation_gate("gitea.read")
if read_block:
return {
"success": False,
"acquired": 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,
"acquired": False,
"reasons": comment_block,
"permission_report": _permission_block_report("gitea.pr.comment"),
}
verify_preflight_purity(remote)
h, o, r = _resolve(remote, host, org, repo)
auth = _auth(h)
profile = get_profile()
identity = _authenticated_username(remote) or profile.get("username") or ""
sid = (session_id or "").strip() or reviewer_pr_lease.new_session_id()
repo_label = f"{o}/{r}"
comments = _fetch_pr_comments(
pr_number, remote=remote, host=host, org=org, repo=repo)
assessment = reviewer_pr_lease.assess_acquire_lease(
comments,
pr_number=pr_number,
reviewer_identity=identity,
profile=profile.get("profile_name") or "unknown",
session_id=sid,
repo=repo_label,
issue_number=issue_number,
worktree=worktree,
candidate_head=candidate_head,
target_branch=target_branch,
target_branch_sha=target_branch_sha,
)
if not assessment.get("acquire_allowed"):
return {
"success": False,
"acquired": False,
"reasons": assessment.get("reasons") or [],
"existing_lease": assessment.get("existing_lease"),
}
body = assessment["lease_body"]
comment_url = f"{repo_api_url(h, o, r)}/issues/{pr_number}/comments"
with _audited(
"comment_pr",
host=h,
remote=remote,
org=o,
repo=r,
pr_number=pr_number,
request_metadata={"source": "acquire_reviewer_pr_lease"},
):
posted = api_request("POST", comment_url, auth, {"body": body})
session_lease = reviewer_pr_lease.record_session_lease({
"pr_number": pr_number,
"issue_number": issue_number,
"session_id": sid,
"reviewer_identity": identity,
"profile": profile.get("profile_name"),
"worktree": worktree,
"phase": "claimed",
"candidate_head": candidate_head,
"target_branch": target_branch,
"target_branch_sha": target_branch_sha,
"repo": repo_label,
"comment_id": posted.get("id"),
})
return {
"success": True,
"acquired": True,
"pr_number": pr_number,
"session_id": sid,
"comment_id": posted.get("id"),
"session_lease": session_lease,
"reasons": [],
}
@mcp.tool()
def gitea_heartbeat_reviewer_pr_lease(
pr_number: int,
phase: str,
worktree: str | None = None,
candidate_head: str | None = None,
target_branch_sha: str | None = None,
remote: str = "dadeschools",
host: str | None = None,
org: str | None = None,
repo: str | None = None,
) -> dict:
"""Post a reviewer lease heartbeat / phase update on the PR thread (#407)."""
comment_block = _profile_operation_gate("gitea.pr.comment")
if comment_block:
return {
"success": False,
"posted": False,
"reasons": comment_block,
"permission_report": _permission_block_report("gitea.pr.comment"),
}
session = reviewer_pr_lease.get_session_lease()
if not session or session.get("pr_number") != pr_number:
return {
"success": False,
"posted": False,
"reasons": [
f"no in-session lease for PR #{pr_number}; acquire first "
"(fail closed)"
],
}
verify_preflight_purity(remote)
h, o, r = _resolve(remote, host, org, repo)
auth = _auth(h)
body = reviewer_pr_lease.format_lease_body(
repo=f"{o}/{r}",
pr_number=pr_number,
issue_number=session.get("issue_number"),
reviewer_identity=session.get("reviewer_identity") or "",
profile=session.get("profile") or "unknown",
session_id=session.get("session_id") or reviewer_pr_lease.new_session_id(),
worktree=worktree or session.get("worktree") or "",
phase=phase,
candidate_head=candidate_head or session.get("candidate_head"),
target_branch=session.get("target_branch") or "master",
target_branch_sha=target_branch_sha or session.get("target_branch_sha"),
)
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": "heartbeat_reviewer_pr_lease", "phase": phase},
):
posted = api_request("POST", comment_url, auth, {"body": body})
updated = reviewer_pr_lease.record_session_lease({
**session,
"phase": phase,
"worktree": worktree or session.get("worktree"),
"candidate_head": candidate_head or session.get("candidate_head"),
"target_branch_sha": target_branch_sha or session.get("target_branch_sha"),
"last_comment_id": posted.get("id"),
})
return {
"success": True,
"posted": True,
"pr_number": pr_number,
"phase": phase,
"comment_id": posted.get("id"),
"session_lease": updated,
"reasons": [],
}
@mcp.tool()
def gitea_assess_reviewer_pr_lease(
pr_number: int,
remote: str = "dadeschools",
host: str | None = None,
org: str | None = None,
repo: str | None = None,
) -> dict:
"""Read-only: assess active reviewer lease state for a PR (#407)."""
read_block = _profile_operation_gate("gitea.read")
if read_block:
return {
"success": False,
"reasons": read_block,
"permission_report": _permission_block_report("gitea.read"),
}
comments = _fetch_pr_comments(
pr_number, remote=remote, host=host, org=org, repo=repo)
active = reviewer_pr_lease.find_active_reviewer_lease(
comments, pr_number=pr_number)
return {
"success": True,
"pr_number": pr_number,
"active_lease": active,
"session_lease": reviewer_pr_lease.get_session_lease(),
"reasons": [],
}
@mcp.tool()
def gitea_list_issue_comments(
issue_number: int,