sysadmin
add87b5708
Treat inventory mergeable/head as advisory. Add classification helper and MCP tool so author sessions re-pin the live PR head before creating a conflict-fix worktree, skip stale inventory false-negatives, and prove classification in final reports.
267 lines
9.4 KiB
Python
267 lines
9.4 KiB
Python
"""Live PR head re-pin before conflict-fix classification (#522).
|
|
|
|
Open PR inventory fields (mergeable, head_sha) can be stale. Author sessions
|
|
must re-fetch live PR state and pin the live head before classifying a PR as
|
|
conflicted or creating a conflict-fix worktree.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from typing import Any
|
|
|
|
_FULL_SHA = re.compile(r"^[0-9a-f]{40}$", re.IGNORECASE)
|
|
|
|
_INVENTORY_HEAD_RE = re.compile(
|
|
r"(?:inventory head sha|stale inventory head sha)\s*:\s*([0-9a-f]{40})",
|
|
re.IGNORECASE,
|
|
)
|
|
_LIVE_HEAD_RE = re.compile(
|
|
r"(?:live head sha|pinned head sha|live pr head sha)\s*:\s*([0-9a-f]{40})",
|
|
re.IGNORECASE,
|
|
)
|
|
_REPINS_TOOL_RE = re.compile(
|
|
r"gitea_assess_conflict_fix_classification",
|
|
re.IGNORECASE,
|
|
)
|
|
_CLASSIFICATION_RE = re.compile(
|
|
r"conflict[- ]fix classification\s*:\s*(\S+)",
|
|
re.IGNORECASE,
|
|
)
|
|
|
|
CLASSIFICATION_STALE_INVENTORY_SKIP = "stale_inventory_skip"
|
|
CLASSIFICATION_CONFLICT_FIX_NEEDED = "conflict_fix_needed"
|
|
CLASSIFICATION_LIVE_MERGEABLE = "live_mergeable_skip"
|
|
CLASSIFICATION_INCOMPLETE = "incomplete_live_repin"
|
|
|
|
_VALID_CLASSIFICATIONS = frozenset({
|
|
CLASSIFICATION_STALE_INVENTORY_SKIP,
|
|
CLASSIFICATION_CONFLICT_FIX_NEEDED,
|
|
CLASSIFICATION_LIVE_MERGEABLE,
|
|
CLASSIFICATION_INCOMPLETE,
|
|
})
|
|
|
|
|
|
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 assess_conflict_fix_classification(
|
|
*,
|
|
pr_number: int,
|
|
inventory_head_sha: str | None = None,
|
|
inventory_mergeable: bool | None = None,
|
|
live_head_sha: str | None = None,
|
|
live_mergeable: bool | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Classify whether conflict-fix work is justified from live PR state.
|
|
|
|
Inventory fields are advisory only. Live head + live mergeable are required
|
|
before any conflict-fix worktree may be created.
|
|
"""
|
|
inv_head = _normalize_sha(inventory_head_sha)
|
|
live_head = _normalize_sha(live_head_sha)
|
|
reasons: list[str] = []
|
|
|
|
if not isinstance(pr_number, int) or pr_number <= 0:
|
|
return {
|
|
"classification": CLASSIFICATION_INCOMPLETE,
|
|
"worktree_allowed": False,
|
|
"skip_author_mutation": True,
|
|
"inventory_stale_head": False,
|
|
"inventory_stale_mergeable": False,
|
|
"pinned_head_sha": None,
|
|
"inventory_head_sha": inv_head,
|
|
"live_head_sha": live_head,
|
|
"inventory_mergeable": inventory_mergeable,
|
|
"live_mergeable": live_mergeable,
|
|
"pr_number": pr_number,
|
|
"reasons": ["pr_number must be a positive integer (fail closed)"],
|
|
}
|
|
|
|
if live_head is None:
|
|
reasons.append(
|
|
"live PR head SHA missing or not a full 40-char hex SHA; "
|
|
"re-fetch the PR before conflict-fix classification (fail closed)"
|
|
)
|
|
if live_mergeable is None:
|
|
reasons.append(
|
|
"live PR mergeable value missing; re-fetch the PR before "
|
|
"conflict-fix classification (fail closed)"
|
|
)
|
|
|
|
if reasons:
|
|
return {
|
|
"classification": CLASSIFICATION_INCOMPLETE,
|
|
"worktree_allowed": False,
|
|
"skip_author_mutation": True,
|
|
"inventory_stale_head": bool(
|
|
inv_head and live_head and inv_head != live_head
|
|
),
|
|
"inventory_stale_mergeable": (
|
|
inventory_mergeable is not None
|
|
and live_mergeable is not None
|
|
and bool(inventory_mergeable) != bool(live_mergeable)
|
|
),
|
|
"pinned_head_sha": live_head,
|
|
"inventory_head_sha": inv_head,
|
|
"live_head_sha": live_head,
|
|
"inventory_mergeable": inventory_mergeable,
|
|
"live_mergeable": live_mergeable,
|
|
"pr_number": pr_number,
|
|
"reasons": reasons,
|
|
}
|
|
|
|
inventory_stale_head = bool(inv_head and inv_head != live_head)
|
|
inventory_stale_mergeable = (
|
|
inventory_mergeable is not None
|
|
and bool(inventory_mergeable) != bool(live_mergeable)
|
|
)
|
|
|
|
# Live mergeable wins: do not start conflict-fix work.
|
|
if live_mergeable is True:
|
|
classification = (
|
|
CLASSIFICATION_STALE_INVENTORY_SKIP
|
|
if (
|
|
inventory_mergeable is False
|
|
or inventory_stale_head
|
|
or inventory_stale_mergeable
|
|
)
|
|
else CLASSIFICATION_LIVE_MERGEABLE
|
|
)
|
|
note = []
|
|
if inventory_stale_head:
|
|
note.append(
|
|
f"inventory head {inv_head} differs from live head {live_head}; "
|
|
"use live head only"
|
|
)
|
|
if inventory_mergeable is False:
|
|
note.append(
|
|
"inventory reported mergeable:false but live PR is mergeable:true; "
|
|
"skip author conflict-fix mutation"
|
|
)
|
|
return {
|
|
"classification": classification,
|
|
"worktree_allowed": False,
|
|
"skip_author_mutation": True,
|
|
"inventory_stale_head": inventory_stale_head,
|
|
"inventory_stale_mergeable": inventory_stale_mergeable,
|
|
"pinned_head_sha": live_head,
|
|
"inventory_head_sha": inv_head,
|
|
"live_head_sha": live_head,
|
|
"inventory_mergeable": inventory_mergeable,
|
|
"live_mergeable": live_mergeable,
|
|
"pr_number": pr_number,
|
|
"reasons": note,
|
|
}
|
|
|
|
# live_mergeable is False → conflict-fix may proceed on the pinned live head.
|
|
note = []
|
|
if inventory_stale_head:
|
|
note.append(
|
|
f"inventory head {inv_head} differs from live head {live_head}; "
|
|
"pin and use live head only for conflict-fix work"
|
|
)
|
|
if inventory_mergeable is True:
|
|
note.append(
|
|
"inventory reported mergeable:true but live PR is mergeable:false; "
|
|
"trust live mergeable and proceed only with live head pin"
|
|
)
|
|
note.append(
|
|
f"live PR #{pr_number} is mergeable:false at pinned head {live_head}; "
|
|
"conflict-fix worktree allowed for that head only"
|
|
)
|
|
return {
|
|
"classification": CLASSIFICATION_CONFLICT_FIX_NEEDED,
|
|
"worktree_allowed": True,
|
|
"skip_author_mutation": False,
|
|
"inventory_stale_head": inventory_stale_head,
|
|
"inventory_stale_mergeable": inventory_stale_mergeable,
|
|
"pinned_head_sha": live_head,
|
|
"inventory_head_sha": inv_head,
|
|
"live_head_sha": live_head,
|
|
"inventory_mergeable": inventory_mergeable,
|
|
"live_mergeable": live_mergeable,
|
|
"pr_number": pr_number,
|
|
"reasons": note,
|
|
}
|
|
|
|
|
|
def assess_conflict_fix_classification_final_report(
|
|
report_text: str,
|
|
**_kwargs: Any,
|
|
) -> dict[str, Any]:
|
|
"""Require live-head re-pin proof when a report claims conflict-fix work (#522)."""
|
|
text = report_text or ""
|
|
lower = text.lower()
|
|
mentions_conflict_fix = (
|
|
"conflict-fix" in lower
|
|
or "conflict fix" in lower
|
|
or "conflict_fix" in lower
|
|
)
|
|
if not mentions_conflict_fix:
|
|
return {"proven": True, "reasons": [], "applicable": False}
|
|
|
|
reasons: list[str] = []
|
|
if not _REPINS_TOOL_RE.search(text):
|
|
reasons.append(
|
|
"conflict-fix report must cite gitea_assess_conflict_fix_classification "
|
|
"(live head re-pin tool)"
|
|
)
|
|
|
|
live_match = _LIVE_HEAD_RE.search(text)
|
|
if not live_match:
|
|
reasons.append(
|
|
"conflict-fix report must state Live head SHA: <40-char hex> "
|
|
"(or Pinned head SHA / Live PR head SHA)"
|
|
)
|
|
else:
|
|
live_sha = _normalize_sha(live_match.group(1))
|
|
if live_sha is None:
|
|
reasons.append("live head SHA is not a full 40-char hex digest")
|
|
|
|
inv_match = _INVENTORY_HEAD_RE.search(text)
|
|
# Inventory head is optional but recommended when classification is stale skip.
|
|
class_match = _CLASSIFICATION_RE.search(text)
|
|
if not class_match:
|
|
reasons.append(
|
|
"conflict-fix report must state Conflict-fix classification: "
|
|
f"<{'|'.join(sorted(_VALID_CLASSIFICATIONS))}>"
|
|
)
|
|
else:
|
|
classification = class_match.group(1).strip().lower().replace(" ", "_")
|
|
# allow hyphenated forms
|
|
classification = classification.replace("-", "_")
|
|
if classification not in _VALID_CLASSIFICATIONS:
|
|
reasons.append(
|
|
f"unknown conflict-fix classification {class_match.group(1)!r}; "
|
|
f"expected one of {sorted(_VALID_CLASSIFICATIONS)}"
|
|
)
|
|
|
|
if inv_match and live_match:
|
|
inv_sha = _normalize_sha(inv_match.group(1))
|
|
live_sha = _normalize_sha(live_match.group(1))
|
|
if inv_sha and live_sha and inv_sha != live_sha:
|
|
# Require explicit note that live head was used.
|
|
if "use live head" not in lower and "live head only" not in lower:
|
|
reasons.append(
|
|
"inventory head differs from live head; report must state that "
|
|
"the live head was used exclusively"
|
|
)
|
|
|
|
return {
|
|
"proven": not reasons,
|
|
"reasons": reasons,
|
|
"applicable": True,
|
|
"inventory_head_sha": _normalize_sha(inv_match.group(1)) if inv_match else None,
|
|
"live_head_sha": _normalize_sha(live_match.group(1)) if live_match else None,
|
|
"classification": (
|
|
class_match.group(1).strip().lower().replace("-", "_").replace(" ", "_")
|
|
if class_match
|
|
else None
|
|
),
|
|
}
|