diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 73dae00..448db8d 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -1134,6 +1134,7 @@ import review_merge_state_machine # noqa: E402 import pr_work_lease # noqa: E402 import workflow_skill # noqa: E402 import conflict_fix_classification # noqa: E402 +import pr_sync_status # noqa: E402 # PR sync / update-by-merge lifecycle import native_mcp_preference # noqa: E402 import branch_cleanup_guard # noqa: E402 import thread_state_ledger_validator # noqa: E402 @@ -13195,6 +13196,616 @@ def gitea_assess_conflict_fix_classification( return result +def _count_commits_behind( + base_url: str, + auth: dict, + *, + pr_head_sha: str, + base_head_sha: str, +) -> int | None: + """Return how many base commits are missing from the PR head, if knowable.""" + # compare old...new returns commits reachable from new not from old. + # pr_head...base_head ≈ commits on base not in PR (behind count). + try: + cmp_url = ( + f"{base_url}/compare/{pr_head_sha}...{base_head_sha}" + ) + data = api_request("GET", cmp_url, auth) or {} + total = data.get("total_commits") + if isinstance(total, int) and total >= 0: + return total + commits = data.get("commits") + if isinstance(commits, list): + return len(commits) + except Exception: + pass + return None + + +def _branch_protection_requires_current_base( + base_url: str, + auth: dict, + *, + base_branch: str, +) -> bool | None: + """Read Gitea branch protection ``block_on_outdated_branch`` when present.""" + branch = (base_branch or "").strip() + if not branch: + return None + try: + # Prefer the named protection rule matching the base branch. + protections = api_request( + "GET", f"{base_url}/branch_protections", auth + ) or [] + if isinstance(protections, list): + for rule in protections: + if not isinstance(rule, dict): + continue + name = (rule.get("branch_name") or rule.get("rule_name") or "").strip() + # Exact match or glob-ish contains for common patterns. + if name == branch or name in ("*", f"{branch}"): + if "block_on_outdated_branch" in rule: + return bool(rule.get("block_on_outdated_branch")) + # Fallback: any protection that mentions the base branch. + for rule in protections: + if not isinstance(rule, dict): + continue + name = (rule.get("branch_name") or rule.get("rule_name") or "").strip() + if branch in name or name.endswith(branch): + if "block_on_outdated_branch" in rule: + return bool(rule.get("block_on_outdated_branch")) + # Branch payload may embed effective protection. + br = api_request("GET", f"{base_url}/branches/{branch}", auth) or {} + prot = br.get("protection") or br.get("effective_branch_protection") or {} + if isinstance(prot, dict) and "block_on_outdated_branch" in prot: + return bool(prot.get("block_on_outdated_branch")) + except Exception: + return None + return None + + +@mcp.tool() +def gitea_assess_pr_sync_status( + pr_number: int, + remote: str = "dadeschools", + host: str | None = None, + org: str | None = None, + repo: str | None = None, + prepared_verdict_head_sha: str | None = None, + branch_protection_requires_current_base: bool | None = None, + checks_status: str | None = None, +) -> dict: + """Read-only: assess whether an approved/open PR is merge-ready or needs sync. + + Distinguishes ``merge_now``, ``update_branch_by_merge``, + ``author_conflict_remediation``, ``fresh_review_required``, and ``blocked``. + + Pins exact PR head and live base head. Never mutates. Never returns tokens. + """ + read_block = _profile_operation_gate("gitea.read") + if read_block: + return { + "success": False, + "performed": False, + "recommended_next_action": pr_sync_status.ACTION_BLOCKED, + "reasons": read_block, + "permission_report": _permission_block_report("gitea.read"), + } + + try: + h, o, r = _resolve(remote, host, org, repo) + except Exception as exc: + return { + "success": False, + "performed": False, + "recommended_next_action": pr_sync_status.ACTION_BLOCKED, + "reasons": [f"repository identity resolve failed: {_redact(str(exc))}"], + } + + auth = _auth(h) + if not auth: + return { + "success": False, + "performed": False, + "recommended_next_action": pr_sync_status.ACTION_BLOCKED, + "reasons": ["credentials unavailable (fail closed)"], + "host": h, + "org": o, + "repo": r, + } + + base = repo_api_url(h, o, r) + try: + pr = api_request("GET", f"{base}/pulls/{pr_number}", auth) or {} + except Exception as exc: + return { + "success": False, + "performed": False, + "recommended_next_action": pr_sync_status.ACTION_BLOCKED, + "reasons": [f"PR details could not be retrieved: {_redact(str(exc))}"], + "host": h, + "org": o, + "repo": r, + "pr_number": pr_number, + } + + pr_state = (pr.get("state") or "").strip().lower() or None + head_obj = pr.get("head") or {} + base_obj = pr.get("base") or {} + pr_head_sha = (head_obj.get("sha") or "").strip() or None + source_branch = (head_obj.get("ref") or "").strip() or None + base_branch = (base_obj.get("ref") or "").strip() or "master" + base_head_sha = (base_obj.get("sha") or "").strip() or None + mergeable = pr.get("mergeable") + # Gitea sometimes exposes conflict via mergeable=false only. + has_conflicts = False if mergeable is True else (True if mergeable is False else None) + + # Prefer live base branch tip over the PR's stored base SHA (may lag). + try: + br = api_request("GET", f"{base}/branches/{base_branch}", auth) or {} + live_base = ((br.get("commit") or {}).get("id") or "").strip() + if live_base: + base_head_sha = live_base + except Exception: + pass + + commits_behind = None + if pr_head_sha and base_head_sha: + commits_behind = _count_commits_behind( + base, auth, pr_head_sha=pr_head_sha, base_head_sha=base_head_sha + ) + if commits_behind is None: + # Fail closed on unknown behind-count only when SHAs differ. + commits_behind = 0 if pr_head_sha == base_head_sha else None + + if branch_protection_requires_current_base is None: + branch_protection_requires_current_base = ( + _branch_protection_requires_current_base( + base, auth, base_branch=base_branch + ) + ) + # When protection cannot be read, fail closed by requiring current base + # whenever the PR is behind (safer for protected repos). Callers may pass + # an explicit False when policy is known not to require it. + if branch_protection_requires_current_base is None: + if commits_behind is not None and commits_behind > 0: + branch_protection_requires_current_base = True + else: + branch_protection_requires_current_base = False + + approval_at_current_head = None + try: + feedback = gitea_get_pr_review_feedback( + pr_number=pr_number, remote=remote, host=host, org=org, repo=repo, + ) + if feedback.get("success"): + approval_at_current_head = bool(feedback.get("approval_at_current_head")) + else: + approval_at_current_head = False + except Exception: + approval_at_current_head = None + + # Locks / leases (best-effort; absence is reported as None/False). + active_author_lock = None + try: + lock = issue_lock_store.load_issue_lock( + remote=remote if remote in REMOTES else (h or "unknown"), + org=o, + repo=r, + issue_number=pr_number, + ) + if lock: + active_author_lock = issue_lock_store.is_lease_live(lock) + else: + active_author_lock = False + except Exception: + active_author_lock = None + + active_reviewer_lease = None + active_merger_lease = None + try: + comments = api_request( + "GET", f"{base}/issues/{pr_number}/comments", auth + ) or [] + if isinstance(comments, list): + rev = pr_work_lease.find_active_reviewer_lease( + comments, pr_number=pr_number + ) + active_reviewer_lease = bool(rev) + # Merger lease comments share reviewer_pr_lease session store when present. + try: + sess = reviewer_pr_lease.get_session_lease() or {} + active_merger_lease = bool( + sess.get("active") + and int(sess.get("pr_number") or 0) == int(pr_number) + and (sess.get("phase") or "").startswith("merger") + ) + except Exception: + active_merger_lease = False + except Exception: + pass + + if checks_status is None: + checks_status = "unknown" + # Combined status on PR head when Actions/status API is available. + if pr_head_sha: + try: + st = api_request( + "GET", + f"{base}/commits/{pr_head_sha}/status", + auth, + ) or {} + state = (st.get("state") or "").strip().lower() + if state: + checks_status = state + elif not st: + checks_status = "none" + except Exception: + checks_status = "unknown" + + assessment = pr_sync_status.assess_pr_sync_status( + host=h, + org=o, + repo=r, + pr_number=pr_number, + pr_state=pr_state, + source_branch=source_branch, + pr_head_sha=pr_head_sha, + base_head_sha=base_head_sha, + commits_behind=commits_behind, + mergeable=mergeable, + has_conflicts=has_conflicts, + branch_protection_requires_current_base=branch_protection_requires_current_base, + approval_at_current_head=approval_at_current_head, + checks_status=checks_status, + active_author_lock=active_author_lock, + active_reviewer_lease=active_reviewer_lease, + active_merger_lease=active_merger_lease, + prepared_verdict_head_sha=prepared_verdict_head_sha, + ) + assessment["remote"] = remote if remote in REMOTES else None + assessment["base_branch"] = base_branch + assessment["success"] = True + assessment["performed"] = False + return assessment + + +@mcp.tool() +def gitea_update_pr_branch_by_merge( + pr_number: int, + expected_pr_head_sha: str, + expected_base_head_sha: str, + remote: str = "dadeschools", + host: str | None = None, + org: str | None = None, + repo: str | None = None, + worktree_path: str | None = None, +) -> dict: + """Author-only: merge live base into the PR branch (Gitea Update-by-merge). + + Pins both expected PR head and expected base head; fails closed on either + race. Never rebases or force-pushes. On conflicts, returns a structured + author-remediation handoff without partial remote mutation. + + Requires author profile, existing issue/PR lock, PR worktree under + branches/, clean control checkout, and master parity. + """ + profile = get_profile() + allowed = profile.get("allowed_operations") or [] + forbidden = profile.get("forbidden_operations") or [] + role = _role_kind(allowed, forbidden) + + # Permission: author branch push / PR mutation surface. + push_block = _profile_operation_gate("gitea.branch.push") + if push_block: + return { + "success": False, + "performed": False, + "mutation_allowed": False, + "reasons": push_block, + "permission_report": _permission_block_report("gitea.branch.push"), + "role_kind": role, + } + + if role != "author": + pre = pr_sync_status.assess_update_pr_branch_preflight( + role_kind=role, + expected_pr_head_sha=expected_pr_head_sha, + live_pr_head_sha=expected_pr_head_sha, + expected_base_head_sha=expected_base_head_sha, + live_base_head_sha=expected_base_head_sha, + style="merge", + ) + pre["success"] = False + return pre + + try: + h, o, r = _resolve(remote, host, org, repo) + except Exception as exc: + return { + "success": False, + "performed": False, + "mutation_allowed": False, + "reasons": [f"repository identity resolve failed: {_redact(str(exc))}"], + "role_kind": role, + } + + # Workspace / control / parity gates. + try: + _verify_role_mutation_workspace( + remote, worktree_path=worktree_path, task="push_branch" + ) + except Exception as exc: + return { + "success": False, + "performed": False, + "mutation_allowed": False, + "reasons": [f"workspace gate failed: {_redact(str(exc))}"], + "role_kind": role, + "host": h, + "org": o, + "repo": r, + "pr_number": pr_number, + } + + control_clean = True + try: + porcelain = _get_workspace_porcelain(PROJECT_ROOT) + # Untracked-only dirt is ignored; tracked edits contaminate control. + tracked_dirty = [ + line for line in (porcelain or "").splitlines() + if line.strip() and not line.startswith("??") + ] + control_clean = not bool(tracked_dirty) + except Exception: + control_clean = False + + master_ok = True + try: + stale = _master_parity_block("gitea.branch.push") + master_ok = not bool(stale) + except Exception: + master_ok = False + + wt = (worktree_path or "").strip() or ( + os.environ.get("GITEA_AUTHOR_WORKTREE") + or os.environ.get("GITEA_ACTIVE_WORKTREE") + or "" + ).strip() + + has_lock = False + try: + lock = issue_lock_store.read_session_issue_lock() + if lock and int(lock.get("issue_number") or 0) == int(pr_number): + has_lock = issue_lock_store.is_lease_live(lock) + if not has_lock: + lock2 = issue_lock_store.load_issue_lock( + remote=remote if remote in REMOTES else (h or "unknown"), + org=o, + repo=r, + issue_number=pr_number, + ) + has_lock = bool(lock2 and issue_lock_store.is_lease_live(lock2)) + except Exception: + has_lock = False + + auth = _auth(h) + if not auth: + return { + "success": False, + "performed": False, + "mutation_allowed": False, + "reasons": ["credentials unavailable (fail closed)"], + "role_kind": role, + "host": h, + "org": o, + "repo": r, + "pr_number": pr_number, + } + + base = repo_api_url(h, o, r) + try: + pr = api_request("GET", f"{base}/pulls/{pr_number}", auth) or {} + except Exception as exc: + return { + "success": False, + "performed": False, + "mutation_allowed": False, + "reasons": [f"PR details could not be retrieved: {_redact(str(exc))}"], + "role_kind": role, + "host": h, + "org": o, + "repo": r, + "pr_number": pr_number, + } + + if (pr.get("state") or "").strip().lower() != "open": + return { + "success": False, + "performed": False, + "mutation_allowed": False, + "reasons": [ + f"PR is not open (state={pr.get('state')}); refuse update" + ], + "role_kind": role, + "host": h, + "org": o, + "repo": r, + "pr_number": pr_number, + } + + head_obj = pr.get("head") or {} + base_obj = pr.get("base") or {} + live_pr_head = (head_obj.get("sha") or "").strip() or None + source_branch = (head_obj.get("ref") or "").strip() or None + base_branch = (base_obj.get("ref") or "").strip() or "master" + live_base_head = (base_obj.get("sha") or "").strip() or None + try: + br = api_request("GET", f"{base}/branches/{base_branch}", auth) or {} + tip = ((br.get("commit") or {}).get("id") or "").strip() + if tip: + live_base_head = tip + except Exception: + pass + + mergeable = pr.get("mergeable") + has_conflicts = False if mergeable is True else (True if mergeable is False else None) + + owns_branch = True + if wt and source_branch: + try: + # Best-effort: worktree branch should match PR source branch. + import subprocess as _sp + out = _sp.check_output( + ["git", "-C", wt, "rev-parse", "--abbrev-ref", "HEAD"], + text=True, + stderr=_sp.DEVNULL, + ).strip() + if out and out != source_branch: + owns_branch = False + except Exception: + owns_branch = None # unknown — do not hard-fail solely on probe + + preflight = pr_sync_status.assess_update_pr_branch_preflight( + role_kind=role, + expected_pr_head_sha=expected_pr_head_sha, + live_pr_head_sha=live_pr_head, + expected_base_head_sha=expected_base_head_sha, + live_base_head_sha=live_base_head, + has_conflicts=has_conflicts, + mergeable=mergeable, + style="merge", + has_author_lock=has_lock, + worktree_path=wt or None, + worktree_owns_source_branch=owns_branch, + control_checkout_clean=control_clean, + master_parity_ok=master_ok, + force_push=False, + rebase=False, + ) + preflight["host"] = h + preflight["org"] = o + preflight["repo"] = r + preflight["pr_number"] = pr_number + preflight["source_branch"] = source_branch + preflight["base_branch"] = base_branch + preflight["worktree_path"] = wt or None + preflight["profile_name"] = profile.get("profile_name") + + if not preflight.get("mutation_allowed"): + preflight["success"] = False + preflight["performed"] = False + if preflight.get("author_remediation_handoff"): + preflight["recommended_next_action"] = ( + pr_sync_status.ACTION_AUTHOR_CONFLICT_REMEDIATION + ) + return preflight + + # Native Gitea update: POST .../pulls/{index}/update?style=merge + update_url = f"{base}/pulls/{pr_number}/update?style=merge" + try: + api_request("POST", update_url, auth) + except Exception as exc: + msg = _redact(str(exc)).lower() + conflictish = any( + token in msg + for token in ("conflict", "409", "merge conflict", "not mergeable") + ) + if conflictish: + return { + "success": False, + "performed": False, + "mutation_allowed": False, + "author_remediation_handoff": True, + "recommended_next_action": ( + pr_sync_status.ACTION_AUTHOR_CONFLICT_REMEDIATION + ), + "reasons": [ + "Gitea refused update-by-merge due to conflicts; " + "no partial remote update applied", + _redact(str(exc)), + ], + "host": h, + "org": o, + "repo": r, + "pr_number": pr_number, + "expected_pr_head_sha": expected_pr_head_sha, + "expected_base_head_sha": expected_base_head_sha, + "live_pr_head_sha_before": live_pr_head, + "style": "merge", + "role_kind": role, + } + return { + "success": False, + "performed": False, + "mutation_allowed": False, + "reasons": [ + f"update-by-merge API failed (fail closed): {_redact(str(exc))}" + ], + "host": h, + "org": o, + "repo": r, + "pr_number": pr_number, + "style": "merge", + "role_kind": role, + } + + # Re-read new head after successful update. + new_head = None + try: + pr_after = api_request("GET", f"{base}/pulls/{pr_number}", auth) or {} + new_head = ((pr_after.get("head") or {}).get("sha") or "").strip() or None + mergeable_after = pr_after.get("mergeable") + except Exception: + pr_after = {} + mergeable_after = None + + transition = pr_sync_status.assess_post_update_head_transition( + former_pr_head_sha=live_pr_head, + new_pr_head_sha=new_head, + former_approval_head_sha=live_pr_head, + prepared_verdict_head_sha=live_pr_head, + ) + + return { + "success": True, + "performed": True, + "mutation_allowed": True, + "style": "merge", + "force_push": False, + "rebase": False, + "host": h, + "org": o, + "repo": r, + "pr_number": pr_number, + "source_branch": source_branch, + "base_branch": base_branch, + "former_pr_head_sha": live_pr_head, + "new_pr_head_sha": new_head, + "base_head_sha": live_base_head, + "mergeable_after": mergeable_after, + "approval_invalidated_for_former_head": transition.get( + "approval_invalidated", True + ), + "reviewer_lease_superseded": transition.get("reviewer_lease_superseded"), + "merger_lease_superseded": transition.get("merger_lease_superseded"), + "prepared_verdict_invalidated": transition.get( + "prepared_verdict_invalidated" + ), + "recommended_next_action": transition.get( + "recommended_next_action", + pr_sync_status.ACTION_FRESH_REVIEW_REQUIRED, + ), + "transition": transition, + "role_kind": role, + "profile_name": profile.get("profile_name"), + "worktree_path": wt or None, + "reasons": list(transition.get("reasons") or []) + [ + "update-by-merge completed via native Gitea API (style=merge only)" + ], + } + + @mcp.tool() def gitea_assess_conflict_fix_push( pr_number: int, @@ -13867,6 +14478,8 @@ def gitea_resolve_task_capability( "commit_files", "gitea_commit_files", "address_pr_change_requests", + "update_pr_branch_by_merge", + "gitea_update_pr_branch_by_merge", "delete_branch", "cleanup_merged_pr_branch", "reconciliation_cleanup", diff --git a/pr_sync_status.py b/pr_sync_status.py new file mode 100644 index 0000000..546c42c --- /dev/null +++ b/pr_sync_status.py @@ -0,0 +1,648 @@ +"""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, + } diff --git a/skills/llm-project-workflow/templates/merge-pr.md b/skills/llm-project-workflow/templates/merge-pr.md index 04b4730..5507096 100644 --- a/skills/llm-project-workflow/templates/merge-pr.md +++ b/skills/llm-project-workflow/templates/merge-pr.md @@ -33,22 +33,34 @@ Steps: *If the current identity does not match the required role (or is the PR author), STOP. Relaunch/switch to the correct profile first.* 2. Verify authenticated identity + active profile. 3. Confirm PR #: author (not you), state open, mergeable, review approved. Check if PR body uses `Closes #N` or `Fixes #N`; if it uses `Implements #N` or `Refs #N`, manual closing will be needed in step 29. -4. Capability evidence (#179): cite the exact gitea_resolve_task_capability +4. **PR sync assess (required):** call `gitea_assess_pr_sync_status` with + explicit `remote`/`org`/`repo`. Route on `recommended_next_action`: + - `merge_now` → continue merger path (do **not** update the branch) + - `update_branch_by_merge` → STOP; hand off to **author** for + `gitea_update_pr_branch_by_merge` (pins expected PR head + base head) + - `author_conflict_remediation` → STOP; author worktree conflict fix + - `fresh_review_required` → STOP; independent re-review at current head + - `blocked` → STOP and diagnose + Never merge on a former-head approval after update/remediation. +5. Capability evidence (#179): cite the exact gitea_resolve_task_capability output (or runtime context) proving merge_pr is allowed — a bare "capability checks passed" claim is downgraded. -5. Final live-state recheck (#179), immediately before the merge mutation — +6. Final live-state recheck (#179), immediately before the merge mutation — re-read the live PR and prove: - PR still open - live head SHA still equals the pinned/reviewed head SHA - base branch unchanged - no undismissed REQUEST_CHANGES / blocking review state remains If any recheck fails → STOP, re-pin, re-validate. -6. If any gate fails → STOP and report. -7. Merge with explicit confirmation (e.g. confirmation="MERGE PR "), +7. If any gate fails → STOP and report. +8. Merge with explicit confirmation (e.g. confirmation="MERGE PR "), pinning the reviewed head SHA (expected_head_sha) and, where supported, the changed-file set. -8. Confirm remote master now contains the merge commit (or the expected changes if squash merged). +9. Confirm remote master now contains the merge commit (or the expected changes if squash merged). *Note: Gitea PR "closed" state is NOT equivalent to "merged". Do not assume a closed PR succeeded without verifying the actual landed changes.* +10. **Sequential queue:** after merge, refresh live `master`, run post-merge + cleanup handoff, then reassess the **next** PR (do not batch-update all + open PRs). Post-merge cleanup (#517): merger sessions must NOT perform ad hoc cleanup. - Record merge mutations separately from cleanup mutations in the controller handoff. diff --git a/skills/llm-project-workflow/workflows/review-merge-pr.md b/skills/llm-project-workflow/workflows/review-merge-pr.md index 95d0bc4..4ac6cce 100644 --- a/skills/llm-project-workflow/workflows/review-merge-pr.md +++ b/skills/llm-project-workflow/workflows/review-merge-pr.md @@ -950,9 +950,53 @@ Final reports must state: * final live head SHA before merge * whether any push occurred during validation +## 26D. PR synchronization and conflict-remediation lifecycle + +**Do not treat “approved” as the final readiness state.** For every approved +open PR, call `gitea_assess_pr_sync_status` (native MCP only) and route by +`recommended_next_action`: + +| Action | Meaning | Next role / tools | +|--------|---------|-------------------| +| `merge_now` | Valid approval at exact current head; conflict-free; update not required; checks ok | Merger: sanctioned merge workflow only — **do not** update the branch | +| `update_branch_by_merge` | Behind live base; protection requires current base; Gitea can merge base without conflicts | **Author only:** `gitea_update_pr_branch_by_merge` with pinned `expected_pr_head_sha` + `expected_base_head_sha` | +| `author_conflict_remediation` | Conflicts / not auto-updatable | **Author only:** existing `branches/` worktree, issue lock, merge master, resolve, test, push; never force-push/rebase | +| `fresh_review_required` | Head changed after approval, or update/remediation produced a new head | Independent reviewer at the **new exact head**; old approvals/leases/verdicts are void | +| `blocked` | Other gate (checks, incomplete facts, closed PR, …) | Diagnose; do not merge or update | + +### Hard rules + +* Pin both expected PR head and expected base head for every update. Either + race → fail closed with no partial mutation. +* Never rebase or force-push author branches. +* Never update an author branch from a reviewer or merger profile. +* Never preserve approval, reviewer lease, merger lease, or prepared verdict + across a head change. +* Process PRs **sequentially**: synchronize/remediate one → review new head → + merge → refresh live `master` → reassess the next PR. Do not update every + open PR at once (each merge restales the rest). +* Conflict remediation only in the existing dedicated issue/PR worktree under + `branches/`. Do not delete that worktree until the updated PR is merged and + cleanup eligibility is proven. +* Post-merge: canonical cleanup handoff to reconciler (section 28). + +### After `gitea_update_pr_branch_by_merge` succeeds + +1. Treat former-head approval as invalidated. +2. Release/supersede obsolete reviewer and merger leases. +3. Route `recommended_next_action=fresh_review_required` at the new head. +4. Independent re-review, then merger lease/adopt + merge at the new head only. + +All Gitea reads/mutations use native MCP. Never substitute direct API, curl, +tea/gh, Web UI mutation by an LLM, database changes, raw Git push, or token +access. + ## 27. Merge rules -Before merge, rerun fresh live checks: +Before merge, call `gitea_assess_pr_sync_status` when the PR is approved/open +and may be behind base. Only proceed with merge when +`recommended_next_action` is `merge_now` and approval remains at the current +head. Then rerun fresh live checks: * whoami * active profile/runtime diff --git a/task_capability_map.py b/task_capability_map.py index 02826af..e2b60c2 100644 --- a/task_capability_map.py +++ b/task_capability_map.py @@ -64,6 +64,24 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = { "permission": "gitea.branch.push", "role": "author", }, + # PR synchronization lifecycle: assess is read-only (any role with gitea.read); + # update-by-merge is author-only and mutates the PR head via Gitea API. + "assess_pr_sync_status": { + "permission": "gitea.read", + "role": "author", + }, + "gitea_assess_pr_sync_status": { + "permission": "gitea.read", + "role": "author", + }, + "update_pr_branch_by_merge": { + "permission": "gitea.branch.push", + "role": "author", + }, + "gitea_update_pr_branch_by_merge": { + "permission": "gitea.branch.push", + "role": "author", + }, "review_pr": { "permission": "gitea.pr.review", "role": "reviewer", diff --git a/tests/test_pr_sync_status.py b/tests/test_pr_sync_status.py new file mode 100644 index 0000000..02a2e31 --- /dev/null +++ b/tests/test_pr_sync_status.py @@ -0,0 +1,338 @@ +"""Hermetic tests for PR synchronization / conflict-remediation lifecycle.""" + +from __future__ import annotations + +import unittest + +from pr_sync_status import ( + ACTION_AUTHOR_CONFLICT_REMEDIATION, + ACTION_BLOCKED, + ACTION_FRESH_REVIEW_REQUIRED, + ACTION_MERGE_NOW, + ACTION_UPDATE_BRANCH_BY_MERGE, + assess_post_update_head_transition, + assess_pr_sync_status, + assess_sequential_queue_step, + assess_update_pr_branch_preflight, + lease_usable_at_head, + merge_approval_usable_at_head, +) + + +def _sha(prefix: str) -> str: + return (prefix + "0" * 40)[:40] + + +PR_HEAD = _sha("aaaaaaaa") +BASE_HEAD = _sha("bbbbbbbb") +NEW_HEAD = _sha("cccccccc") +OLD_HEAD = _sha("dddddddd") + + +def _base_kwargs(**overrides): + data = { + "host": "gitea.prgs.cc", + "org": "Scaled-Tech-Consulting", + "repo": "Gitea-Tools", + "pr_number": 100, + "pr_state": "open", + "source_branch": "fix/issue-100", + "pr_head_sha": PR_HEAD, + "base_head_sha": BASE_HEAD, + "commits_behind": 0, + "mergeable": True, + "has_conflicts": False, + "branch_protection_requires_current_base": False, + "approval_at_current_head": True, + "checks_status": "success", + "active_author_lock": True, + "active_reviewer_lease": False, + "active_merger_lease": False, + } + data.update(overrides) + return data + + +class TestAssessPrSyncStatus(unittest.TestCase): + def test_approved_current_mergeable_merge_now(self): + result = assess_pr_sync_status(**_base_kwargs()) + self.assertEqual(result["recommended_next_action"], ACTION_MERGE_NOW) + self.assertTrue(result["approval_valid_for_merge"]) + self.assertFalse(result["stale_approval"]) + self.assertEqual(result["pr_head_sha"], PR_HEAD) + self.assertEqual(result["base_head_sha"], BASE_HEAD) + + def test_approved_outdated_update_not_required_merge_now(self): + result = assess_pr_sync_status( + **_base_kwargs( + commits_behind=3, + branch_protection_requires_current_base=False, + ) + ) + self.assertEqual(result["recommended_next_action"], ACTION_MERGE_NOW) + self.assertTrue(result["approval_valid_for_merge"]) + self.assertTrue(any("does not require" in r for r in result["reasons"])) + + def test_outdated_update_required_no_conflict(self): + result = assess_pr_sync_status( + **_base_kwargs( + commits_behind=2, + branch_protection_requires_current_base=True, + mergeable=True, + has_conflicts=False, + ) + ) + self.assertEqual( + result["recommended_next_action"], ACTION_UPDATE_BRANCH_BY_MERGE + ) + self.assertFalse(result["approval_valid_for_merge"]) + self.assertTrue(any("update_branch_by_merge" in r for r in result["reasons"])) + + def test_outdated_has_conflicts_author_remediation(self): + result = assess_pr_sync_status( + **_base_kwargs( + commits_behind=5, + branch_protection_requires_current_base=True, + mergeable=False, + has_conflicts=True, + ) + ) + self.assertEqual( + result["recommended_next_action"], ACTION_AUTHOR_CONFLICT_REMEDIATION + ) + self.assertFalse(result["approval_valid_for_merge"]) + + def test_stale_approval_fresh_review_required(self): + result = assess_pr_sync_status( + **_base_kwargs( + approval_at_current_head=False, + commits_behind=0, + ) + ) + self.assertEqual( + result["recommended_next_action"], ACTION_FRESH_REVIEW_REQUIRED + ) + self.assertTrue(result["stale_approval"]) + self.assertFalse(result["approval_valid_for_merge"]) + + def test_stale_prepared_verdict_cannot_cross_heads(self): + result = assess_pr_sync_status( + **_base_kwargs( + approval_at_current_head=True, + prepared_verdict_head_sha=OLD_HEAD, + ) + ) + self.assertEqual( + result["recommended_next_action"], ACTION_FRESH_REVIEW_REQUIRED + ) + self.assertTrue(result["stale_prepared_verdict"]) + self.assertFalse(result["approval_valid_for_merge"]) + + def test_missing_heads_fail_closed(self): + result = assess_pr_sync_status( + **_base_kwargs(pr_head_sha="short", base_head_sha=None) + ) + self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED) + self.assertTrue(any("PR head" in r for r in result["reasons"])) + + def test_closed_pr_blocked(self): + result = assess_pr_sync_status(**_base_kwargs(pr_state="closed")) + self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED) + + def test_failed_checks_block_merge_now(self): + result = assess_pr_sync_status(**_base_kwargs(checks_status="failure")) + self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED) + + def test_stale_approval_with_update_required_routes_update(self): + result = assess_pr_sync_status( + **_base_kwargs( + approval_at_current_head=False, + commits_behind=1, + branch_protection_requires_current_base=True, + mergeable=True, + has_conflicts=False, + ) + ) + self.assertEqual( + result["recommended_next_action"], ACTION_UPDATE_BRANCH_BY_MERGE + ) + + +class TestUpdatePrBranchPreflight(unittest.TestCase): + def _ok_kwargs(self, **overrides): + data = { + "role_kind": "author", + "expected_pr_head_sha": PR_HEAD, + "live_pr_head_sha": PR_HEAD, + "expected_base_head_sha": BASE_HEAD, + "live_base_head_sha": BASE_HEAD, + "has_conflicts": False, + "mergeable": True, + "style": "merge", + "has_author_lock": True, + "worktree_path": "/Users/x/Development/Gitea-Tools/branches/issue-100", + "worktree_owns_source_branch": True, + "control_checkout_clean": True, + "master_parity_ok": True, + "force_push": False, + "rebase": False, + } + data.update(overrides) + return data + + def test_author_allowed_when_preflight_clean(self): + result = assess_update_pr_branch_preflight(**self._ok_kwargs()) + self.assertTrue(result["mutation_allowed"]) + self.assertFalse(result["head_race"]) + self.assertFalse(result["base_race"]) + + def test_head_race_fail_closed(self): + result = assess_update_pr_branch_preflight( + **self._ok_kwargs(live_pr_head_sha=NEW_HEAD) + ) + self.assertFalse(result["mutation_allowed"]) + self.assertTrue(result["head_race"]) + self.assertTrue(any("PR head race" in r for r in result["reasons"])) + + def test_base_race_fail_closed(self): + result = assess_update_pr_branch_preflight( + **self._ok_kwargs(live_base_head_sha=NEW_HEAD) + ) + self.assertFalse(result["mutation_allowed"]) + self.assertTrue(result["base_race"]) + self.assertTrue(any("base head race" in r for r in result["reasons"])) + + def test_conflicts_structured_handoff_no_mutation(self): + result = assess_update_pr_branch_preflight( + **self._ok_kwargs(has_conflicts=True, mergeable=False) + ) + self.assertFalse(result["mutation_allowed"]) + self.assertTrue(result["author_remediation_handoff"]) + self.assertTrue(any("conflicts" in r.lower() for r in result["reasons"])) + + def test_reviewer_denied(self): + result = assess_update_pr_branch_preflight( + **self._ok_kwargs(role_kind="reviewer") + ) + self.assertFalse(result["mutation_allowed"]) + self.assertTrue(any("author-only" in r for r in result["reasons"])) + + def test_merger_denied(self): + result = assess_update_pr_branch_preflight( + **self._ok_kwargs(role_kind="merger") + ) + self.assertFalse(result["mutation_allowed"]) + self.assertTrue(any("author-only" in r for r in result["reasons"])) + + def test_no_force_push(self): + result = assess_update_pr_branch_preflight( + **self._ok_kwargs(force_push=True) + ) + self.assertFalse(result["mutation_allowed"]) + self.assertTrue(any("force-push" in r for r in result["reasons"])) + + def test_no_rebase(self): + result = assess_update_pr_branch_preflight( + **self._ok_kwargs(style="rebase", rebase=True) + ) + self.assertFalse(result["mutation_allowed"]) + self.assertTrue(any("rebase" in r for r in result["reasons"])) + + def test_missing_lock_fail_closed(self): + result = assess_update_pr_branch_preflight( + **self._ok_kwargs(has_author_lock=False) + ) + self.assertFalse(result["mutation_allowed"]) + + def test_worktree_not_under_branches(self): + result = assess_update_pr_branch_preflight( + **self._ok_kwargs(worktree_path="/tmp/not-a-branches-path") + ) + self.assertFalse(result["mutation_allowed"]) + self.assertTrue(any("branches/" in r for r in result["reasons"])) + + +class TestPostUpdateHeadTransition(unittest.TestCase): + def test_update_invalidates_approval_and_requires_fresh_review(self): + result = assess_post_update_head_transition( + former_pr_head_sha=PR_HEAD, + new_pr_head_sha=NEW_HEAD, + former_approval_head_sha=PR_HEAD, + former_reviewer_lease_head_sha=PR_HEAD, + former_merger_lease_head_sha=PR_HEAD, + prepared_verdict_head_sha=PR_HEAD, + ) + self.assertTrue(result["head_changed"]) + self.assertTrue(result["approval_invalidated"]) + self.assertTrue(result["reviewer_lease_superseded"]) + self.assertTrue(result["merger_lease_superseded"]) + self.assertTrue(result["prepared_verdict_invalidated"]) + self.assertEqual( + result["recommended_next_action"], ACTION_FRESH_REVIEW_REQUIRED + ) + + def test_same_head_unexpected(self): + result = assess_post_update_head_transition( + former_pr_head_sha=PR_HEAD, + new_pr_head_sha=PR_HEAD, + ) + self.assertFalse(result["head_changed"]) + self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED) + + +class TestStaleArtifacts(unittest.TestCase): + def test_stale_approval_cannot_authorize_merge(self): + result = merge_approval_usable_at_head( + current_head_sha=NEW_HEAD, + approved_head_sha=PR_HEAD, + ) + self.assertFalse(result["approval_usable"]) + + def test_fresh_approval_usable(self): + result = merge_approval_usable_at_head( + current_head_sha=NEW_HEAD, + approved_head_sha=NEW_HEAD, + ) + self.assertTrue(result["approval_usable"]) + + def test_stale_reviewer_lease_cannot_cross_heads(self): + result = lease_usable_at_head( + lease_kind="reviewer", + lease_head_sha=PR_HEAD, + current_head_sha=NEW_HEAD, + ) + self.assertFalse(result["lease_usable"]) + + def test_stale_merger_lease_cannot_cross_heads(self): + result = lease_usable_at_head( + lease_kind="merger", + lease_head_sha=PR_HEAD, + current_head_sha=NEW_HEAD, + ) + self.assertFalse(result["lease_usable"]) + + +class TestSequentialQueue(unittest.TestCase): + def test_reassess_after_merge_and_master_refresh(self): + result = assess_sequential_queue_step( + just_merged=True, + master_refreshed=True, + next_pr_number=101, + ) + self.assertTrue(result["reassess_allowed"]) + self.assertTrue(result["process_next"]) + self.assertTrue(result["post_merge_cleanup_handoff"]) + self.assertEqual(result["next_pr_number"], 101) + + def test_block_reassess_without_master_refresh(self): + result = assess_sequential_queue_step( + just_merged=True, + master_refreshed=False, + next_pr_number=101, + ) + self.assertFalse(result["reassess_allowed"]) + self.assertTrue(any("master must be refreshed" in r for r in result["reasons"])) + + +if __name__ == "__main__": + unittest.main()