sysadmin
ebb1512ec9
The conflict-resolution rebase incorrectly kept ISSUE_LOCK_FILE in the merged_cleanup_reconcile import, but that symbol was moved to issue_lock_provenance on master. This broke webui import collection. Fixes the reviewer-reported regression on PR #448.
331 lines
11 KiB
Python
331 lines
11 KiB
Python
"""Lease and collision visibility for the web UI (#433)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import re
|
|
import subprocess
|
|
from dataclasses import dataclass
|
|
from typing import Any, Callable
|
|
from urllib.parse import urlparse
|
|
|
|
from gitea_auth import api_fetch_page, get_auth_header, repo_api_url
|
|
from issue_claim_heartbeat import build_claim_inventory
|
|
from issue_lock_provenance import ISSUE_LOCK_FILE
|
|
from merged_cleanup_reconcile import read_issue_lock
|
|
|
|
from webui.project_registry import ProjectRecord, load_registry
|
|
from webui.queue_loader import _extract_linked_issue, _fetch_issues, _fetch_prs
|
|
|
|
_REVIEWER_LEASE_MARKER = "<!-- mcp-review-lease:v1 -->"
|
|
_REVIEWER_FIELD_RE = re.compile(
|
|
r"^\s*([a-z_]+)\s*:\s*(.+?)\s*$",
|
|
re.IGNORECASE | re.MULTILINE,
|
|
)
|
|
_ISSUE_BRANCH_RE = re.compile(r"issue-(\d+)", re.IGNORECASE)
|
|
|
|
_COLLISION_HISTORY = (
|
|
{"number": 267, "title": "Author work leases"},
|
|
{"number": 268, "title": "Issue claim heartbeat leases"},
|
|
{"number": 400, "title": "Early duplicate-work detection"},
|
|
{"number": 407, "title": "Per-PR reviewer leases"},
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class CollisionWarning:
|
|
kind: str
|
|
message: str
|
|
issue_number: int | None = None
|
|
pr_numbers: tuple[int, ...] = ()
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class LeaseSnapshot:
|
|
project_id: str
|
|
repo_label: str
|
|
issue_lock: dict[str, Any] | None
|
|
claim_inventory: dict[str, Any]
|
|
reviewer_leases: tuple[dict[str, Any], ...]
|
|
duplicate_prs: tuple[CollisionWarning, ...]
|
|
duplicate_branches: tuple[CollisionWarning, ...]
|
|
collision_history: tuple[dict[str, Any], ...]
|
|
fetch_error: str | None = None
|
|
|
|
|
|
def _repo_root() -> str:
|
|
override = (os.environ.get("WEBUI_REPO_ROOT") or "").strip()
|
|
if override:
|
|
return os.path.realpath(override)
|
|
return os.path.realpath(os.path.join(os.path.dirname(__file__), ".."))
|
|
|
|
|
|
def _host_from_url(remote_host: str) -> str:
|
|
parsed = urlparse(remote_host.strip())
|
|
return parsed.netloc or remote_host.strip().rstrip("/")
|
|
|
|
|
|
def _parse_pr_ref(value: str | None) -> int | None:
|
|
digits = re.sub(r"[^\d]", "", value or "")
|
|
return int(digits) if digits.isdigit() else None
|
|
|
|
|
|
def parse_reviewer_lease_comment(body: str) -> dict[str, Any] | None:
|
|
text = body or ""
|
|
if _REVIEWER_LEASE_MARKER not in text:
|
|
return None
|
|
fields: dict[str, str] = {}
|
|
for match in _REVIEWER_FIELD_RE.finditer(text):
|
|
fields[match.group(1).strip().lower()] = match.group(2).strip()
|
|
if not fields:
|
|
return None
|
|
return {
|
|
"pr_number": _parse_pr_ref(fields.get("pr")),
|
|
"issue_number": _parse_pr_ref(fields.get("issue")),
|
|
"reviewer_identity": fields.get("reviewer_identity"),
|
|
"profile": fields.get("profile"),
|
|
"phase": (fields.get("phase") or "").strip().lower() or None,
|
|
"expires_at": fields.get("expires_at"),
|
|
"blocker": fields.get("blocker"),
|
|
}
|
|
|
|
|
|
def _fetch_comments(
|
|
host: str,
|
|
org: str,
|
|
repo: str,
|
|
auth: str,
|
|
*,
|
|
issue_number: int,
|
|
) -> list[dict]:
|
|
url = f"{repo_api_url(host, org, repo)}/issues/{issue_number}/comments"
|
|
comments: list[dict] = []
|
|
page = 1
|
|
while page <= 10:
|
|
raw_page, meta = api_fetch_page(url, auth, page=page, limit=50)
|
|
comments.extend(raw_page)
|
|
if meta.get("is_final_page"):
|
|
break
|
|
page += 1
|
|
return comments
|
|
|
|
|
|
def _list_local_branch_names(project_root: str) -> list[str]:
|
|
result = subprocess.run(
|
|
["git", "-C", project_root, "branch", "--list", "--format=%(refname:short)"],
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
if result.returncode != 0:
|
|
return []
|
|
return [line.strip() for line in (result.stdout or "").splitlines() if line.strip()]
|
|
|
|
|
|
def _duplicate_pr_warnings(raw_prs: list[dict]) -> list[CollisionWarning]:
|
|
issue_to_prs: dict[int, list[int]] = {}
|
|
for pr in raw_prs:
|
|
linked = _extract_linked_issue(pr.get("title"), pr.get("body"))
|
|
if linked is None:
|
|
continue
|
|
issue_to_prs.setdefault(linked, []).append(int(pr["number"]))
|
|
warnings: list[CollisionWarning] = []
|
|
for issue_number, pr_numbers in sorted(issue_to_prs.items()):
|
|
if len(pr_numbers) > 1:
|
|
warnings.append(
|
|
CollisionWarning(
|
|
kind="duplicate-pr",
|
|
issue_number=issue_number,
|
|
pr_numbers=tuple(sorted(pr_numbers)),
|
|
message=(
|
|
f"Issue #{issue_number} has {len(pr_numbers)} open PRs: "
|
|
f"{', '.join(f'#{n}' for n in sorted(pr_numbers))} (#400)"
|
|
),
|
|
)
|
|
)
|
|
return warnings
|
|
|
|
|
|
def _duplicate_branch_warnings(branch_names: list[str]) -> list[CollisionWarning]:
|
|
issue_to_branches: dict[int, list[str]] = {}
|
|
for name in branch_names:
|
|
match = _ISSUE_BRANCH_RE.search(name)
|
|
if not match:
|
|
continue
|
|
issue_num = int(match.group(1))
|
|
issue_to_branches.setdefault(issue_num, []).append(name)
|
|
warnings: list[CollisionWarning] = []
|
|
for issue_number, branches in sorted(issue_to_branches.items()):
|
|
if len(branches) > 1:
|
|
warnings.append(
|
|
CollisionWarning(
|
|
kind="duplicate-branch",
|
|
issue_number=issue_number,
|
|
message=(
|
|
f"Issue #{issue_number} has {len(branches)} local branches: "
|
|
f"{', '.join(branches)}"
|
|
),
|
|
)
|
|
)
|
|
return warnings
|
|
|
|
|
|
def _extract_reviewer_leases(
|
|
raw_prs: list[dict],
|
|
*,
|
|
fetch_comments: Callable[[int], list[dict]],
|
|
) -> list[dict[str, Any]]:
|
|
leases: list[dict[str, Any]] = []
|
|
for pr in raw_prs:
|
|
pr_number = int(pr["number"])
|
|
for comment in fetch_comments(pr_number):
|
|
parsed = parse_reviewer_lease_comment(comment.get("body") or "")
|
|
if not parsed:
|
|
continue
|
|
leases.append(
|
|
{
|
|
**parsed,
|
|
"pr_number": parsed.get("pr_number") or pr_number,
|
|
"comment_id": comment.get("id"),
|
|
"author": (comment.get("user") or {}).get("login"),
|
|
"created_at": comment.get("created_at"),
|
|
}
|
|
)
|
|
active_phases = {"claimed", "validating", "approved", "request-changes", "merging"}
|
|
return [lease for lease in leases if (lease.get("phase") or "") in active_phases]
|
|
|
|
|
|
def load_lease_snapshot(
|
|
*,
|
|
project_id: str | None = None,
|
|
project_root: str | None = None,
|
|
issue_lock_path: str | None = None,
|
|
fetch_prs: Callable | None = None,
|
|
fetch_issues: Callable | None = None,
|
|
fetch_comments: Callable[[str, str, str, str, int], list[dict]] | None = None,
|
|
) -> LeaseSnapshot:
|
|
registry = load_registry()
|
|
project: ProjectRecord | None = None
|
|
if project_id:
|
|
project = next((p for p in registry.projects if p.id == project_id), None)
|
|
else:
|
|
project = registry.projects[0] if registry.projects else None
|
|
|
|
root = project_root or _repo_root()
|
|
lock = read_issue_lock(issue_lock_path if issue_lock_path is not None else ISSUE_LOCK_FILE)
|
|
|
|
if project is None:
|
|
return LeaseSnapshot(
|
|
project_id=project_id or "",
|
|
repo_label="",
|
|
issue_lock=lock,
|
|
claim_inventory={"entries": [], "counts": {}},
|
|
reviewer_leases=(),
|
|
duplicate_prs=(),
|
|
duplicate_branches=(),
|
|
collision_history=_COLLISION_HISTORY,
|
|
fetch_error="project not found in registry",
|
|
)
|
|
|
|
host = _host_from_url(project.remote_host)
|
|
pr_fetch = fetch_prs or _fetch_prs
|
|
issue_fetch = fetch_issues or _fetch_issues
|
|
comment_fetch = fetch_comments or _fetch_comments
|
|
using_live = fetch_prs is None or fetch_issues is None
|
|
auth = get_auth_header(host) if using_live else "test-auth"
|
|
|
|
if using_live and not auth:
|
|
return LeaseSnapshot(
|
|
project_id=project.id,
|
|
repo_label=f"{project.gitea_owner}/{project.repo_name}",
|
|
issue_lock=lock,
|
|
claim_inventory={"entries": [], "counts": {}},
|
|
reviewer_leases=(),
|
|
duplicate_prs=(),
|
|
duplicate_branches=_duplicate_branch_warnings(_list_local_branch_names(root)),
|
|
collision_history=_COLLISION_HISTORY,
|
|
fetch_error=(
|
|
f"Gitea credentials unavailable for {host}; "
|
|
"remote lease artifacts cannot be loaded"
|
|
),
|
|
)
|
|
|
|
try:
|
|
raw_prs, _ = pr_fetch(host, project.gitea_owner, project.repo_name, auth)
|
|
raw_issues, _ = issue_fetch(host, project.gitea_owner, project.repo_name, auth)
|
|
except Exception as exc: # noqa: BLE001
|
|
return LeaseSnapshot(
|
|
project_id=project.id,
|
|
repo_label=f"{project.gitea_owner}/{project.repo_name}",
|
|
issue_lock=lock,
|
|
claim_inventory={"entries": [], "counts": {}},
|
|
reviewer_leases=(),
|
|
duplicate_prs=(),
|
|
duplicate_branches=_duplicate_branch_warnings(_list_local_branch_names(root)),
|
|
collision_history=_COLLISION_HISTORY,
|
|
fetch_error=f"Gitea fetch failed: {exc}",
|
|
)
|
|
|
|
comments_by_issue: dict[int, list[dict]] = {}
|
|
for issue in raw_issues:
|
|
if not any(lb.get("name") == "status:in-progress" for lb in issue.get("labels", [])):
|
|
continue
|
|
number = int(issue["number"])
|
|
comments_by_issue[number] = comment_fetch(
|
|
host, project.gitea_owner, project.repo_name, auth, issue_number=number
|
|
)
|
|
|
|
branch_names = _list_local_branch_names(root)
|
|
inventory = build_claim_inventory(
|
|
issues=raw_issues,
|
|
comments_by_issue=comments_by_issue,
|
|
open_prs=raw_prs,
|
|
branch_names=branch_names,
|
|
)
|
|
|
|
def _pr_comments(pr_number: int) -> list[dict]:
|
|
return comment_fetch(
|
|
host, project.gitea_owner, project.repo_name, auth, issue_number=pr_number
|
|
)
|
|
|
|
reviewer_leases = tuple(_extract_reviewer_leases(raw_prs, fetch_comments=_pr_comments))
|
|
|
|
return LeaseSnapshot(
|
|
project_id=project.id,
|
|
repo_label=f"{project.gitea_owner}/{project.repo_name}",
|
|
issue_lock=lock,
|
|
claim_inventory=inventory,
|
|
reviewer_leases=reviewer_leases,
|
|
duplicate_prs=tuple(_duplicate_pr_warnings(raw_prs)),
|
|
duplicate_branches=tuple(_duplicate_branch_warnings(branch_names)),
|
|
collision_history=_COLLISION_HISTORY,
|
|
)
|
|
|
|
|
|
def snapshot_to_dict(snapshot: LeaseSnapshot) -> dict[str, Any]:
|
|
return {
|
|
"project_id": snapshot.project_id,
|
|
"repo_label": snapshot.repo_label,
|
|
"issue_lock": snapshot.issue_lock,
|
|
"claim_inventory": snapshot.claim_inventory,
|
|
"reviewer_leases": list(snapshot.reviewer_leases),
|
|
"duplicate_prs": [
|
|
{
|
|
"kind": w.kind,
|
|
"message": w.message,
|
|
"issue_number": w.issue_number,
|
|
"pr_numbers": list(w.pr_numbers),
|
|
}
|
|
for w in snapshot.duplicate_prs
|
|
],
|
|
"duplicate_branches": [
|
|
{
|
|
"kind": w.kind,
|
|
"message": w.message,
|
|
"issue_number": w.issue_number,
|
|
}
|
|
for w in snapshot.duplicate_branches
|
|
],
|
|
"collision_history": list(snapshot.collision_history),
|
|
"fetch_error": snapshot.fetch_error,
|
|
} |