sysadmin
98fbfb3de7
Prevent approved PRs from stalling when master advances. Add gitea_assess_pr_sync_status and gitea_update_pr_branch_by_merge with head/base pinning, author-only updates, conflict handoff, and approval invalidation after head changes. Wire task capability map, sequential controller routing in review-merge workflow, and hermetic AC tests.
649 lines
25 KiB
Python
649 lines
25 KiB
Python
"""PR synchronization and conflict-remediation lifecycle (#PR-SYNC).
|
|
|
|
Pure assessment helpers for:
|
|
|
|
* ``gitea_assess_pr_sync_status`` — recommend the next sanctioned action for an
|
|
open PR when the base branch advances, approvals go stale, or conflicts appear.
|
|
* ``gitea_update_pr_branch_by_merge`` preflight — author-only, fail-closed pin
|
|
of both PR head and live base head; never rebase/force-push.
|
|
|
|
All recommendation logic is hermetic (no network) so unit tests cover every
|
|
acceptance path without Gitea credentials.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from typing import Any
|
|
|
|
_FULL_SHA = re.compile(r"^[0-9a-f]{40}$", re.IGNORECASE)
|
|
|
|
# Recommended next actions (controller / skill vocabulary).
|
|
ACTION_MERGE_NOW = "merge_now"
|
|
ACTION_UPDATE_BRANCH_BY_MERGE = "update_branch_by_merge"
|
|
ACTION_AUTHOR_CONFLICT_REMEDIATION = "author_conflict_remediation"
|
|
ACTION_FRESH_REVIEW_REQUIRED = "fresh_review_required"
|
|
ACTION_BLOCKED = "blocked"
|
|
|
|
_VALID_ACTIONS = frozenset({
|
|
ACTION_MERGE_NOW,
|
|
ACTION_UPDATE_BRANCH_BY_MERGE,
|
|
ACTION_AUTHOR_CONFLICT_REMEDIATION,
|
|
ACTION_FRESH_REVIEW_REQUIRED,
|
|
ACTION_BLOCKED,
|
|
})
|
|
|
|
# Author-only mutation; reviewer/merger must never update author branches.
|
|
_AUTHOR_UPDATE_ROLES = frozenset({"author"})
|
|
_DENIED_UPDATE_ROLES = frozenset({"reviewer", "merger", "reconciler", "mixed", "limited"})
|
|
|
|
# Allowed Gitea update style — merge only (never rebase).
|
|
UPDATE_STYLE_MERGE = "merge"
|
|
_FORBIDDEN_UPDATE_STYLES = frozenset({"rebase", "rebase-merge", "squash", "force"})
|
|
|
|
|
|
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 _normalize_role(role: str | None) -> str:
|
|
return (role or "").strip().lower() or "unknown"
|
|
|
|
|
|
def assess_pr_sync_status(
|
|
*,
|
|
host: str | None = None,
|
|
org: str | None = None,
|
|
repo: str | None = None,
|
|
pr_number: int,
|
|
pr_state: str | None = None,
|
|
source_branch: str | None = None,
|
|
pr_head_sha: str | None = None,
|
|
base_head_sha: str | None = None,
|
|
commits_behind: int | None = None,
|
|
mergeable: bool | None = None,
|
|
has_conflicts: bool | None = None,
|
|
branch_protection_requires_current_base: bool | None = None,
|
|
approval_at_current_head: bool | None = None,
|
|
checks_status: str | None = None,
|
|
active_author_lock: bool | None = None,
|
|
active_reviewer_lease: bool | None = None,
|
|
active_merger_lease: bool | None = None,
|
|
prepared_verdict_head_sha: str | None = None,
|
|
checks_required: bool = True,
|
|
) -> dict[str, Any]:
|
|
"""Recommend the next sanctioned PR lifecycle action.
|
|
|
|
Decision order (fail closed):
|
|
|
|
1. Incomplete identity / open state / head SHAs → ``blocked``
|
|
2. Conflicts (or mergeable false with conflict signal) →
|
|
``author_conflict_remediation``
|
|
3. Head changed after approval / prepared verdict for former head →
|
|
``fresh_review_required`` (unless conflicts already routed author work)
|
|
4. Behind live base + branch protection requires current base +
|
|
auto-mergeable → ``update_branch_by_merge``
|
|
5. Approval at current head + mergeable + checks ok (+ update not
|
|
required) → ``merge_now``
|
|
6. Otherwise ``blocked`` with structured reasons
|
|
"""
|
|
reasons: list[str] = []
|
|
pr_head = _normalize_sha(pr_head_sha)
|
|
base_head = _normalize_sha(base_head_sha)
|
|
prepared_head = _normalize_sha(prepared_verdict_head_sha)
|
|
state = (pr_state or "").strip().lower()
|
|
checks = (checks_status or "unknown").strip().lower()
|
|
behind = commits_behind if isinstance(commits_behind, int) and commits_behind >= 0 else None
|
|
requires_current = bool(branch_protection_requires_current_base)
|
|
approval_ok = bool(approval_at_current_head)
|
|
|
|
# Derive conflict when caller only supplies mergeable=False.
|
|
if has_conflicts is None and mergeable is False:
|
|
has_conflicts = True
|
|
conflicts = bool(has_conflicts) if has_conflicts is not None else None
|
|
|
|
result: dict[str, Any] = {
|
|
"host": (host or "").strip() or None,
|
|
"org": (org or "").strip() or None,
|
|
"repo": (repo or "").strip() or None,
|
|
"pr_number": pr_number,
|
|
"pr_state": state or None,
|
|
"source_branch": (source_branch or "").strip() or None,
|
|
"pr_head_sha": pr_head,
|
|
"base_head_sha": base_head,
|
|
"commits_behind": behind,
|
|
"mergeable": mergeable,
|
|
"has_conflicts": conflicts,
|
|
"branch_protection_requires_current_base": requires_current,
|
|
"approval_at_current_head": approval_ok if approval_at_current_head is not None else None,
|
|
"checks_status": checks,
|
|
"active_locks_and_leases": {
|
|
"author_lock": bool(active_author_lock) if active_author_lock is not None else None,
|
|
"reviewer_lease": bool(active_reviewer_lease) if active_reviewer_lease is not None else None,
|
|
"merger_lease": bool(active_merger_lease) if active_merger_lease is not None else None,
|
|
},
|
|
"prepared_verdict_head_sha": prepared_head,
|
|
"recommended_next_action": ACTION_BLOCKED,
|
|
"approval_valid_for_merge": False,
|
|
"stale_approval": False,
|
|
"stale_prepared_verdict": False,
|
|
"reasons": reasons,
|
|
"success": True,
|
|
"performed": False,
|
|
}
|
|
|
|
if not isinstance(pr_number, int) or pr_number <= 0:
|
|
reasons.append("pr_number must be a positive integer (fail closed)")
|
|
return result
|
|
|
|
if state and state not in ("open",):
|
|
reasons.append(f"PR is not open (state={state}); cannot synchronize or merge")
|
|
result["recommended_next_action"] = ACTION_BLOCKED
|
|
return result
|
|
if not state:
|
|
reasons.append("PR state missing (fail closed)")
|
|
return result
|
|
|
|
if pr_head is None:
|
|
reasons.append(
|
|
"exact PR head SHA missing or not a full 40-char hex SHA (fail closed)"
|
|
)
|
|
if base_head is None:
|
|
reasons.append(
|
|
"exact live base head SHA missing or not a full 40-char hex SHA (fail closed)"
|
|
)
|
|
if behind is None:
|
|
reasons.append("commits_behind missing or invalid (fail closed)")
|
|
if mergeable is None:
|
|
reasons.append("mergeable signal missing (fail closed)")
|
|
if approval_at_current_head is None:
|
|
reasons.append("approval_at_current_head missing (fail closed)")
|
|
if branch_protection_requires_current_base is None:
|
|
reasons.append(
|
|
"branch_protection_requires_current_base missing (fail closed)"
|
|
)
|
|
|
|
if reasons:
|
|
result["recommended_next_action"] = ACTION_BLOCKED
|
|
return result
|
|
|
|
# Stale prepared verdict / approval across heads.
|
|
stale_prepared = bool(prepared_head and prepared_head != pr_head)
|
|
result["stale_prepared_verdict"] = stale_prepared
|
|
stale_approval = not approval_ok
|
|
result["stale_approval"] = stale_approval
|
|
result["approval_valid_for_merge"] = bool(approval_ok and not stale_prepared)
|
|
|
|
# ── Conflicts: author remediation in dedicated worktree ──────────────
|
|
if conflicts is True or mergeable is False:
|
|
if conflicts is True or mergeable is False:
|
|
# Distinguish pure non-mergeable without explicit conflict flag
|
|
# still routes author remediation (Gitea cannot auto-update).
|
|
reasons.append(
|
|
f"PR #{pr_number} has conflicts or is not mergeable at head "
|
|
f"{pr_head}; route author conflict remediation in the existing "
|
|
"issue/PR worktree under branches/ (never force-push or rebase)"
|
|
)
|
|
if stale_approval:
|
|
reasons.append(
|
|
"any approval at a former head is invalid; after remediation "
|
|
"require a fresh independent review at the new exact head"
|
|
)
|
|
result["recommended_next_action"] = ACTION_AUTHOR_CONFLICT_REMEDIATION
|
|
result["approval_valid_for_merge"] = False
|
|
return result
|
|
|
|
# ── Stale approval / prepared verdict at former head ─────────────────
|
|
if not approval_ok or stale_prepared:
|
|
if behind and behind > 0 and requires_current and mergeable is True:
|
|
# Must update first; update will also invalidate approval.
|
|
reasons.append(
|
|
f"PR is {behind} commit(s) behind live base and branch protection "
|
|
"requires current base; automatic merge-from-base is available"
|
|
)
|
|
reasons.append(
|
|
"approval is not valid at the current head (or prepared verdict "
|
|
"is head-scoped to a former SHA); after update require fresh review"
|
|
)
|
|
result["recommended_next_action"] = ACTION_UPDATE_BRANCH_BY_MERGE
|
|
result["approval_valid_for_merge"] = False
|
|
return result
|
|
reasons.append(
|
|
"approval is not valid at the exact current PR head "
|
|
f"({pr_head}); never reuse a former-head approval for merge"
|
|
)
|
|
if stale_prepared:
|
|
reasons.append(
|
|
f"prepared verdict pinned to former head {prepared_head} cannot "
|
|
f"authorize review/merge at current head {pr_head}"
|
|
)
|
|
if active_reviewer_lease:
|
|
reasons.append(
|
|
"active reviewer lease must be released or superseded for the "
|
|
"new head before a fresh independent review"
|
|
)
|
|
if active_merger_lease:
|
|
reasons.append(
|
|
"active merger lease cannot cross heads; release or re-acquire "
|
|
"at the new exact head"
|
|
)
|
|
result["recommended_next_action"] = ACTION_FRESH_REVIEW_REQUIRED
|
|
result["approval_valid_for_merge"] = False
|
|
return result
|
|
|
|
# ── Outdated + protection requires current base, auto-updatable ──────
|
|
if behind and behind > 0 and requires_current:
|
|
if mergeable is True and conflicts is not True:
|
|
reasons.append(
|
|
f"PR is {behind} commit(s) behind live base {base_head}; "
|
|
"branch protection requires the PR branch to contain current base; "
|
|
"Gitea can merge base into the PR branch without conflicts"
|
|
)
|
|
reasons.append(
|
|
"route author-only update_branch_by_merge; never rebase or force-push; "
|
|
"new head invalidates prior approval and requires fresh independent review"
|
|
)
|
|
result["recommended_next_action"] = ACTION_UPDATE_BRANCH_BY_MERGE
|
|
# Approval today is at old head that will change — not mergeable yet
|
|
# for final merge until re-reviewed, but assess marks update path.
|
|
result["approval_valid_for_merge"] = False
|
|
result["stale_approval"] = True # will become stale after update
|
|
return result
|
|
reasons.append(
|
|
"update required by branch protection but automatic merge is not available"
|
|
)
|
|
result["recommended_next_action"] = ACTION_BLOCKED
|
|
return result
|
|
|
|
# ── Checks gate for merge_now ────────────────────────────────────────
|
|
if checks_required and checks not in ("success", "passed", "ok", "none", "skipped", "not_required"):
|
|
if checks in ("pending", "running", "queued"):
|
|
reasons.append(f"required checks are not finished (status={checks})")
|
|
result["recommended_next_action"] = ACTION_BLOCKED
|
|
return result
|
|
if checks in ("failure", "failed", "error", "cancelled"):
|
|
reasons.append(f"required checks failed (status={checks})")
|
|
result["recommended_next_action"] = ACTION_BLOCKED
|
|
return result
|
|
# unknown — fail closed when checks_required
|
|
if checks == "unknown":
|
|
reasons.append("checks status unknown (fail closed)")
|
|
result["recommended_next_action"] = ACTION_BLOCKED
|
|
return result
|
|
|
|
# ── Ready to merge without update ────────────────────────────────────
|
|
# Includes: current with approval; outdated when update is NOT required.
|
|
if approval_ok and mergeable is True and conflicts is not True:
|
|
if behind and behind > 0 and not requires_current:
|
|
reasons.append(
|
|
f"PR is {behind} commit(s) behind live base but branch protection "
|
|
"does not require the latest base; preserve the approved head"
|
|
)
|
|
else:
|
|
reasons.append(
|
|
"PR has a valid approval at the exact current head, is conflict-free "
|
|
"and mergeable; do not update the branch unnecessarily"
|
|
)
|
|
reasons.append("route directly to the sanctioned merger workflow (merge_now)")
|
|
result["recommended_next_action"] = ACTION_MERGE_NOW
|
|
result["approval_valid_for_merge"] = True
|
|
return result
|
|
|
|
reasons.append("no sanctioned next action matched the observed PR state (fail closed)")
|
|
result["recommended_next_action"] = ACTION_BLOCKED
|
|
return result
|
|
|
|
|
|
def assess_update_pr_branch_preflight(
|
|
*,
|
|
role_kind: str | None,
|
|
expected_pr_head_sha: str | None,
|
|
live_pr_head_sha: str | None,
|
|
expected_base_head_sha: str | None,
|
|
live_base_head_sha: str | None,
|
|
has_conflicts: bool | None = None,
|
|
mergeable: bool | None = None,
|
|
style: str | None = UPDATE_STYLE_MERGE,
|
|
has_author_lock: bool | None = None,
|
|
worktree_path: str | None = None,
|
|
worktree_owns_source_branch: bool | None = None,
|
|
control_checkout_clean: bool | None = None,
|
|
master_parity_ok: bool | None = None,
|
|
force_push: bool = False,
|
|
rebase: bool = False,
|
|
) -> dict[str, Any]:
|
|
"""Author-only preflight for update-by-merge; fail closed on any race.
|
|
|
|
When conflicts exist, returns a structured author-remediation handoff and
|
|
sets ``mutation_allowed=False`` so no partial remote update is performed.
|
|
"""
|
|
reasons: list[str] = []
|
|
role = _normalize_role(role_kind)
|
|
exp_pr = _normalize_sha(expected_pr_head_sha)
|
|
live_pr = _normalize_sha(live_pr_head_sha)
|
|
exp_base = _normalize_sha(expected_base_head_sha)
|
|
live_base = _normalize_sha(live_base_head_sha)
|
|
style_norm = (style or "").strip().lower() or UPDATE_STYLE_MERGE
|
|
|
|
conflicts = has_conflicts
|
|
if conflicts is None and mergeable is False:
|
|
conflicts = True
|
|
|
|
result: dict[str, Any] = {
|
|
"mutation_allowed": False,
|
|
"performed": False,
|
|
"style": style_norm,
|
|
"role_kind": role,
|
|
"expected_pr_head_sha": exp_pr,
|
|
"live_pr_head_sha": live_pr,
|
|
"expected_base_head_sha": exp_base,
|
|
"live_base_head_sha": live_base,
|
|
"head_race": False,
|
|
"base_race": False,
|
|
"has_conflicts": conflicts,
|
|
"author_remediation_handoff": False,
|
|
"approval_invalidated_for_former_head": False,
|
|
"force_push": bool(force_push),
|
|
"rebase": bool(rebase),
|
|
"reasons": reasons,
|
|
"success": True,
|
|
}
|
|
|
|
# Role gate — author only.
|
|
if role not in _AUTHOR_UPDATE_ROLES:
|
|
reasons.append(
|
|
f"role '{role}' cannot update an author PR branch "
|
|
"(author-only; reviewer/merger denial)"
|
|
)
|
|
return result
|
|
|
|
if force_push:
|
|
reasons.append("force-push is forbidden for PR branch update (fail closed)")
|
|
return result
|
|
if rebase or style_norm in _FORBIDDEN_UPDATE_STYLES:
|
|
reasons.append(
|
|
f"update style '{style_norm}' forbidden; only style=merge is allowed "
|
|
"(never rebase)"
|
|
)
|
|
return result
|
|
if style_norm != UPDATE_STYLE_MERGE:
|
|
reasons.append(
|
|
f"unknown update style '{style_norm}'; expected '{UPDATE_STYLE_MERGE}'"
|
|
)
|
|
return result
|
|
|
|
if not exp_pr or not live_pr:
|
|
reasons.append(
|
|
"expected and live PR head SHAs must be full 40-char hex (fail closed)"
|
|
)
|
|
if not exp_base or not live_base:
|
|
reasons.append(
|
|
"expected and live base head SHAs must be full 40-char hex (fail closed)"
|
|
)
|
|
if reasons:
|
|
return result
|
|
|
|
if exp_pr != live_pr:
|
|
result["head_race"] = True
|
|
reasons.append(
|
|
f"PR head race: expected {exp_pr} but live is {live_pr} "
|
|
"(fail closed; no partial mutation)"
|
|
)
|
|
return result
|
|
if exp_base != live_base:
|
|
result["base_race"] = True
|
|
reasons.append(
|
|
f"base head race: expected {exp_base} but live is {live_base} "
|
|
"(fail closed; no partial mutation)"
|
|
)
|
|
return result
|
|
|
|
if has_author_lock is not True:
|
|
reasons.append(
|
|
"existing issue/PR lock required before update_branch_by_merge (fail closed)"
|
|
)
|
|
wt = (worktree_path or "").strip()
|
|
if not wt:
|
|
reasons.append("existing PR worktree path required under branches/ (fail closed)")
|
|
else:
|
|
norm = wt.replace("\\", "/")
|
|
if "/branches/" not in norm and not norm.startswith("branches/"):
|
|
reasons.append(
|
|
f"worktree_path '{wt}' must be under branches/ (fail closed)"
|
|
)
|
|
if worktree_owns_source_branch is False:
|
|
reasons.append(
|
|
"worktree does not own the PR source branch (fail closed)"
|
|
)
|
|
if control_checkout_clean is False:
|
|
reasons.append("control checkout is not clean (fail closed)")
|
|
if master_parity_ok is False:
|
|
reasons.append("runtime/master parity failed (fail closed)")
|
|
|
|
if conflicts is True:
|
|
result["author_remediation_handoff"] = True
|
|
reasons.append(
|
|
"conflicts present: return structured author-remediation handoff "
|
|
"without creating a partial remote update"
|
|
)
|
|
reasons.append(
|
|
"resolve conflicts explicitly in the existing dedicated worktree, "
|
|
"run focused and regression tests, commit/push via sanctioned author "
|
|
"workflow, then route the new exact head to independent review"
|
|
)
|
|
return result
|
|
|
|
if mergeable is False and conflicts is not False:
|
|
result["author_remediation_handoff"] = True
|
|
reasons.append(
|
|
"PR is not mergeable; refuse automatic update and hand off to "
|
|
"author conflict remediation (fail closed)"
|
|
)
|
|
return result
|
|
|
|
if reasons:
|
|
return result
|
|
|
|
result["mutation_allowed"] = True
|
|
reasons.append(
|
|
"preflight passed: author may merge live base into the PR branch via "
|
|
"native MCP update (style=merge only)"
|
|
)
|
|
return result
|
|
|
|
|
|
def assess_post_update_head_transition(
|
|
*,
|
|
former_pr_head_sha: str | None,
|
|
new_pr_head_sha: str | None,
|
|
former_approval_head_sha: str | None = None,
|
|
former_reviewer_lease_head_sha: str | None = None,
|
|
former_merger_lease_head_sha: str | None = None,
|
|
prepared_verdict_head_sha: str | None = None,
|
|
) -> dict[str, Any]:
|
|
"""After a successful update-by-merge, invalidate cross-head artifacts.
|
|
|
|
The new head never inherits approval, reviewer/merger leases, or prepared
|
|
verdicts from the former head.
|
|
"""
|
|
former = _normalize_sha(former_pr_head_sha)
|
|
new = _normalize_sha(new_pr_head_sha)
|
|
reasons: list[str] = []
|
|
result: dict[str, Any] = {
|
|
"former_pr_head_sha": former,
|
|
"new_pr_head_sha": new,
|
|
"head_changed": bool(former and new and former != new),
|
|
"approval_invalidated": False,
|
|
"reviewer_lease_superseded": False,
|
|
"merger_lease_superseded": False,
|
|
"prepared_verdict_invalidated": False,
|
|
"recommended_next_action": ACTION_BLOCKED,
|
|
"reasons": reasons,
|
|
"success": True,
|
|
}
|
|
|
|
if not former or not new:
|
|
reasons.append("former and new PR head SHAs required (fail closed)")
|
|
return result
|
|
if former == new:
|
|
reasons.append(
|
|
"PR head unchanged after update; unexpected for merge-from-base"
|
|
)
|
|
result["recommended_next_action"] = ACTION_BLOCKED
|
|
return result
|
|
|
|
result["head_changed"] = True
|
|
# Approval at former head is always void at new head.
|
|
former_approval = _normalize_sha(former_approval_head_sha) or former
|
|
if former_approval != new:
|
|
result["approval_invalidated"] = True
|
|
reasons.append(
|
|
f"approval for former head {former_approval} is invalid at new head "
|
|
f"{new}; never preserve approval across a head change"
|
|
)
|
|
|
|
rev_lease = _normalize_sha(former_reviewer_lease_head_sha)
|
|
if rev_lease and rev_lease != new:
|
|
result["reviewer_lease_superseded"] = True
|
|
reasons.append(
|
|
f"reviewer lease for head {rev_lease} cannot cross to {new}; "
|
|
"release or supersede before fresh review"
|
|
)
|
|
elif rev_lease is None:
|
|
# Unknown lease head still must not authorize at new head.
|
|
result["reviewer_lease_superseded"] = True
|
|
reasons.append(
|
|
"any reviewer lease scoped to the former head is superseded at the new head"
|
|
)
|
|
|
|
mer_lease = _normalize_sha(former_merger_lease_head_sha)
|
|
if mer_lease and mer_lease != new:
|
|
result["merger_lease_superseded"] = True
|
|
reasons.append(
|
|
f"merger lease for head {mer_lease} cannot cross to {new}"
|
|
)
|
|
else:
|
|
result["merger_lease_superseded"] = True
|
|
reasons.append(
|
|
"any merger lease scoped to the former head is superseded at the new head"
|
|
)
|
|
|
|
prepared = _normalize_sha(prepared_verdict_head_sha)
|
|
if prepared and prepared != new:
|
|
result["prepared_verdict_invalidated"] = True
|
|
reasons.append(
|
|
f"prepared verdict for head {prepared} cannot authorize the new head {new}"
|
|
)
|
|
elif prepared is None:
|
|
result["prepared_verdict_invalidated"] = True
|
|
reasons.append(
|
|
"prepared verdicts associated with the former head are invalidated"
|
|
)
|
|
|
|
result["recommended_next_action"] = ACTION_FRESH_REVIEW_REQUIRED
|
|
reasons.append(
|
|
"route the new exact head to independent review "
|
|
f"({ACTION_FRESH_REVIEW_REQUIRED})"
|
|
)
|
|
return result
|
|
|
|
|
|
def assess_sequential_queue_step(
|
|
*,
|
|
just_merged: bool,
|
|
master_refreshed: bool,
|
|
next_pr_number: int | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Controller sequential processing: reassess only after merge + master refresh."""
|
|
reasons: list[str] = []
|
|
if just_merged and not master_refreshed:
|
|
reasons.append(
|
|
"master must be refreshed after each merge before reassessing the next PR "
|
|
"(do not update every PR simultaneously)"
|
|
)
|
|
return {
|
|
"reassess_allowed": False,
|
|
"process_next": False,
|
|
"next_pr_number": next_pr_number,
|
|
"reasons": reasons,
|
|
"success": True,
|
|
}
|
|
if not just_merged:
|
|
reasons.append("no merge completed; sequential step does not advance")
|
|
return {
|
|
"reassess_allowed": False,
|
|
"process_next": False,
|
|
"next_pr_number": next_pr_number,
|
|
"reasons": reasons,
|
|
"success": True,
|
|
}
|
|
if next_pr_number:
|
|
reasons.append(
|
|
"merge completed and live master refreshed; reassess the next PR "
|
|
f"#{next_pr_number}"
|
|
)
|
|
else:
|
|
reasons.append(
|
|
"merge completed and live master refreshed; reassess the next PR"
|
|
)
|
|
return {
|
|
"reassess_allowed": True,
|
|
"process_next": True,
|
|
"next_pr_number": next_pr_number,
|
|
"post_merge_cleanup_handoff": True,
|
|
"reasons": reasons,
|
|
"success": True,
|
|
}
|
|
|
|
|
|
def merge_approval_usable_at_head(
|
|
*,
|
|
current_head_sha: str | None,
|
|
approved_head_sha: str | None,
|
|
) -> dict[str, Any]:
|
|
"""Stale approval cannot authorize merge (head-scoped only)."""
|
|
current = _normalize_sha(current_head_sha)
|
|
approved = _normalize_sha(approved_head_sha)
|
|
ok = bool(current and approved and current == approved)
|
|
reasons: list[str] = []
|
|
if not ok:
|
|
reasons.append(
|
|
f"stale approval: approved head '{approved or '(none)'}' does not "
|
|
f"match current head '{current or '(none)'}'; cannot authorize merge"
|
|
)
|
|
return {
|
|
"approval_usable": ok,
|
|
"current_head_sha": current,
|
|
"approved_head_sha": approved,
|
|
"reasons": reasons,
|
|
}
|
|
|
|
|
|
def lease_usable_at_head(
|
|
*,
|
|
lease_kind: str,
|
|
lease_head_sha: str | None,
|
|
current_head_sha: str | None,
|
|
) -> dict[str, Any]:
|
|
"""Reviewer/merger leases cannot cross heads."""
|
|
current = _normalize_sha(current_head_sha)
|
|
lease_head = _normalize_sha(lease_head_sha)
|
|
kind = (lease_kind or "").strip().lower() or "unknown"
|
|
ok = bool(current and lease_head and current == lease_head)
|
|
reasons: list[str] = []
|
|
if not ok:
|
|
reasons.append(
|
|
f"stale {kind} lease: lease head '{lease_head or '(none)'}' cannot "
|
|
f"authorize work at current head '{current or '(none)'}'"
|
|
)
|
|
return {
|
|
"lease_usable": ok,
|
|
"lease_kind": kind,
|
|
"lease_head_sha": lease_head,
|
|
"current_head_sha": current,
|
|
"reasons": reasons,
|
|
}
|