sysadmin
749e480baf
Expose read-only assess/scan tools and capability planning so already-landed open PRs can be reconciled without review or merge. Register the gitea-reconcile-landed-pr skill and workflow-source verification.
292 lines
9.6 KiB
Python
292 lines
9.6 KiB
Python
"""Already-landed PR reconciliation workflow helpers (#301).
|
|
|
|
Read-only assessment and capability planning for reconciling open PRs whose
|
|
heads are already ancestors of the target branch. Does not invoke review or
|
|
merge paths.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import subprocess
|
|
from typing import Any
|
|
|
|
import gitea_config
|
|
from merged_cleanup_reconcile import extract_linked_issue, is_head_ancestor_of_ref
|
|
|
|
ELIGIBILITY_ALREADY_LANDED = "ALREADY_LANDED_RECONCILE_REQUIRED"
|
|
ELIGIBILITY_NOT_LANDED = "NOT_ALREADY_LANDED"
|
|
ELIGIBILITY_STALE_TARGET = "TARGET_BRANCH_UNVERIFIED"
|
|
ELIGIBILITY_PR_NOT_OPEN = "PR_NOT_OPEN"
|
|
|
|
OUTCOME_FULL_RECONCILE = "FULL_RECONCILE_CLOSE_ALLOWED"
|
|
OUTCOME_PARTIAL_COMMENT = "PARTIAL_RECONCILE_COMMENT_THEN_STOP"
|
|
OUTCOME_RECOVERY_HANDOFF = "RECOVERY_HANDOFF_ONLY"
|
|
OUTCOME_NOT_LANDED = "NOT_LANDED_NO_ACTION"
|
|
OUTCOME_GATE_NOT_PROVEN = "GATE_NOT_PROVEN"
|
|
|
|
RECONCILE_WORKFLOW_MARKERS = (
|
|
"workflows/reconcile-landed-pr.md",
|
|
"reconcile-landed-pr.md",
|
|
)
|
|
|
|
RECONCILE_TASK_MARKERS = (
|
|
"reconcile-landed-pr",
|
|
"reconcile already-landed",
|
|
"reconcile_already_landed",
|
|
)
|
|
|
|
|
|
def fetch_target_branch(project_root: str, remote: str, branch: str) -> dict[str, Any]:
|
|
"""Fetch *branch* from *remote* and return the resolved SHA."""
|
|
ref = f"{remote}/{branch}"
|
|
fetch = subprocess.run(
|
|
["git", "-C", project_root, "fetch", remote, branch],
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
if fetch.returncode != 0:
|
|
return {
|
|
"success": False,
|
|
"target_branch": branch,
|
|
"target_ref": ref,
|
|
"target_branch_sha": None,
|
|
"reasons": [
|
|
f"git fetch {remote} {branch} failed: "
|
|
f"{(fetch.stderr or fetch.stdout or '').strip()}"
|
|
],
|
|
}
|
|
|
|
rev = subprocess.run(
|
|
["git", "-C", project_root, "rev-parse", ref],
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
if rev.returncode != 0:
|
|
return {
|
|
"success": False,
|
|
"target_branch": branch,
|
|
"target_ref": ref,
|
|
"target_branch_sha": None,
|
|
"reasons": [
|
|
f"git rev-parse {ref} failed: "
|
|
f"{(rev.stderr or rev.stdout or '').strip()}"
|
|
],
|
|
}
|
|
|
|
return {
|
|
"success": True,
|
|
"target_branch": branch,
|
|
"target_ref": ref,
|
|
"target_branch_sha": (rev.stdout or "").strip(),
|
|
"reasons": [],
|
|
"git_fetch_command": f"git fetch {remote} {branch}",
|
|
}
|
|
|
|
|
|
def profile_reconciliation_capabilities(
|
|
allowed: list[str] | None,
|
|
forbidden: list[str] | None,
|
|
) -> dict[str, bool]:
|
|
"""Map active profile operations to reconciliation capabilities."""
|
|
allowed = allowed or []
|
|
forbidden = forbidden or []
|
|
|
|
def can(op: str) -> bool:
|
|
return gitea_config.check_operation(op, allowed, forbidden)[0]
|
|
|
|
return {
|
|
"read": can("gitea.read"),
|
|
"comment_pr": can("gitea.pr.comment"),
|
|
"comment_issue": can("gitea.issue.comment"),
|
|
"close_pr": can("gitea.pr.close"),
|
|
"close_issue": can("gitea.issue.close"),
|
|
"review_pr": can("gitea.pr.review") or can("gitea.pr.approve"),
|
|
"merge_pr": can("gitea.pr.merge"),
|
|
}
|
|
|
|
|
|
def assess_open_pr_reconciliation(
|
|
*,
|
|
pr: dict[str, Any],
|
|
project_root: str,
|
|
remote: str,
|
|
target_branch: str,
|
|
target_fetch: dict[str, Any] | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Assess whether an open PR is eligible for already-landed reconciliation."""
|
|
pr_number = int(pr.get("number") or 0)
|
|
pr_state = (pr.get("state") or "").strip().lower()
|
|
head = pr.get("head") or {}
|
|
head_sha = head.get("sha") if isinstance(head, dict) else None
|
|
head_ref = head.get("ref") if isinstance(head, dict) else None
|
|
base = pr.get("base") or {}
|
|
base_ref = base.get("ref") if isinstance(base, dict) else None
|
|
title = pr.get("title") or ""
|
|
body = pr.get("body") or ""
|
|
|
|
fetch_result = target_fetch or fetch_target_branch(
|
|
project_root, remote, target_branch
|
|
)
|
|
linked_issue = extract_linked_issue(title, body)
|
|
|
|
result: dict[str, Any] = {
|
|
"pr_number": pr_number,
|
|
"pr_state": pr_state,
|
|
"candidate_head_sha": head_sha,
|
|
"head_ref": head_ref,
|
|
"base_ref": base_ref or target_branch,
|
|
"target_branch": target_branch,
|
|
"target_branch_sha": fetch_result.get("target_branch_sha"),
|
|
"linked_issue": linked_issue,
|
|
"git_ref_mutations": [],
|
|
"reasons": [],
|
|
"review_merge_allowed": False,
|
|
}
|
|
if fetch_result.get("git_fetch_command"):
|
|
result["git_ref_mutations"].append(fetch_result["git_fetch_command"])
|
|
|
|
if pr_state != "open":
|
|
result["eligibility_class"] = ELIGIBILITY_PR_NOT_OPEN
|
|
result["ancestor_proof"] = None
|
|
result["reconciliation_allowed"] = False
|
|
result["reasons"].append(f"PR #{pr_number} state is {pr_state!r}, not open")
|
|
return result
|
|
|
|
if not fetch_result.get("success"):
|
|
result["eligibility_class"] = ELIGIBILITY_STALE_TARGET
|
|
result["ancestor_proof"] = None
|
|
result["reconciliation_allowed"] = False
|
|
result["reasons"].extend(fetch_result.get("reasons") or [])
|
|
return result
|
|
|
|
target_ref = fetch_result.get("target_ref") or f"{remote}/{target_branch}"
|
|
ancestor = is_head_ancestor_of_ref(project_root, head_sha, target_ref)
|
|
result["ancestor_proof"] = ancestor
|
|
|
|
if ancestor is None:
|
|
result["eligibility_class"] = ELIGIBILITY_STALE_TARGET
|
|
result["reconciliation_allowed"] = False
|
|
result["reasons"].append(
|
|
f"ancestor check failed for head {head_sha!r} against {target_ref}"
|
|
)
|
|
return result
|
|
|
|
if ancestor:
|
|
result["eligibility_class"] = ELIGIBILITY_ALREADY_LANDED
|
|
result["reconciliation_allowed"] = True
|
|
return result
|
|
|
|
result["eligibility_class"] = ELIGIBILITY_NOT_LANDED
|
|
result["reconciliation_allowed"] = False
|
|
result["reasons"].append(
|
|
f"PR head {head_sha} is not an ancestor of {target_ref}"
|
|
)
|
|
return result
|
|
|
|
|
|
def resolve_reconciliation_plan(
|
|
*,
|
|
assessment: dict[str, Any],
|
|
capabilities: dict[str, bool] | None,
|
|
) -> dict[str, Any]:
|
|
"""Map eligibility + profile capabilities to a reconciliation outcome."""
|
|
caps = capabilities or {}
|
|
reasons: list[str] = []
|
|
|
|
if not assessment.get("reconciliation_allowed"):
|
|
outcome = OUTCOME_NOT_LANDED
|
|
if assessment.get("eligibility_class") in {
|
|
ELIGIBILITY_STALE_TARGET,
|
|
ELIGIBILITY_PR_NOT_OPEN,
|
|
}:
|
|
outcome = OUTCOME_GATE_NOT_PROVEN
|
|
reasons.extend(assessment.get("reasons") or [])
|
|
return {
|
|
"outcome": outcome,
|
|
"close_pr_allowed": False,
|
|
"comment_pr_allowed": False,
|
|
"close_issue_allowed": False,
|
|
"comment_issue_allowed": False,
|
|
"review_merge_allowed": False,
|
|
"missing_capabilities": _missing_reconciliation_capabilities(caps),
|
|
"safe_next_action": (
|
|
"Do not reconcile via review/merge; repair missing proof first."
|
|
),
|
|
"reasons": reasons,
|
|
}
|
|
|
|
close_pr = bool(caps.get("close_pr"))
|
|
comment_pr = bool(caps.get("comment_pr"))
|
|
close_issue = bool(caps.get("close_issue"))
|
|
comment_issue = bool(caps.get("comment_issue"))
|
|
|
|
if caps.get("review_pr") or caps.get("merge_pr"):
|
|
reasons.append(
|
|
"reconciliation path must not use review/merge capabilities"
|
|
)
|
|
|
|
if close_pr:
|
|
outcome = OUTCOME_FULL_RECONCILE
|
|
safe_next_action = (
|
|
"Run reconciliation close via exact gitea.pr.close after proof; "
|
|
"do not approve or merge."
|
|
)
|
|
elif comment_pr:
|
|
outcome = OUTCOME_PARTIAL_COMMENT
|
|
safe_next_action = (
|
|
"Post one reconciliation comment with ancestor proof, then stop "
|
|
"for an authorized close profile."
|
|
)
|
|
else:
|
|
outcome = OUTCOME_RECOVERY_HANDOFF
|
|
safe_next_action = (
|
|
"Produce a recovery handoff naming missing gitea.pr.close and/or "
|
|
"gitea.pr.comment; do not loop through review/merge."
|
|
)
|
|
reasons.append("gitea.pr.close is not available in the active profile")
|
|
|
|
return {
|
|
"outcome": outcome,
|
|
"close_pr_allowed": close_pr,
|
|
"comment_pr_allowed": comment_pr,
|
|
"close_issue_allowed": close_issue,
|
|
"comment_issue_allowed": comment_issue,
|
|
"review_merge_allowed": False,
|
|
"missing_capabilities": _missing_reconciliation_capabilities(caps),
|
|
"safe_next_action": safe_next_action,
|
|
"reasons": reasons,
|
|
}
|
|
|
|
|
|
def _missing_reconciliation_capabilities(caps: dict[str, bool]) -> list[str]:
|
|
missing = []
|
|
if not caps.get("read"):
|
|
missing.append("gitea.read")
|
|
if not caps.get("close_pr"):
|
|
missing.append("gitea.pr.close")
|
|
if not caps.get("comment_pr"):
|
|
missing.append("gitea.pr.comment")
|
|
return missing
|
|
|
|
|
|
def assess_reconcile_workflow_source(report_text: str) -> dict[str, Any]:
|
|
"""Reconciliation reports must cite the canonical workflow file."""
|
|
lower = (report_text or "").lower()
|
|
reasons = []
|
|
if not any(marker in lower for marker in RECONCILE_WORKFLOW_MARKERS):
|
|
reasons.append(
|
|
"reconciliation report missing workflow source "
|
|
"(workflows/reconcile-landed-pr.md)"
|
|
)
|
|
if not any(marker in lower for marker in RECONCILE_TASK_MARKERS):
|
|
reasons.append(
|
|
"reconciliation report missing task mode declaration "
|
|
"(reconcile-landed-pr)"
|
|
)
|
|
return {
|
|
"complete": not reasons,
|
|
"downgraded": bool(reasons),
|
|
"reasons": reasons,
|
|
} |