"""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, }