"""Per-PR reviewer leases for safe parallel review sessions (#407).""" from __future__ import annotations import os import re import uuid from datetime import datetime, timedelta, timezone from typing import Any MARKER = "" _FIELD_RE = re.compile( r"^\s*([a-z_]+)\s*:\s*(.+?)\s*$", re.IGNORECASE | re.MULTILINE, ) _FULL_SHA = re.compile(r"^[0-9a-f]{40}$", re.IGNORECASE) _TERMINAL_PHASES = frozenset({"done", "released", "blocked"}) _ACTIVE_PHASES = frozenset({ "claimed", "validating", "approved", "request-changes", "merging", }) DEFAULT_LEASE_TTL_MINUTES = 120 STALE_WARNING_MINUTES = 30 RECLAIMABLE_MINUTES = 60 _SESSION_LEASE: dict[str, Any] | None = None def _parse_timestamp(value: str | None) -> datetime | None: if not value: return None text = value.strip() if text.endswith("Z"): text = text[:-1] + "+00:00" try: parsed = datetime.fromisoformat(text) except ValueError: return None if parsed.tzinfo is None: return parsed.replace(tzinfo=timezone.utc) return parsed.astimezone(timezone.utc) def _normalize_sha(value: str | None) -> str | None: text = (value or "").strip().lower() return text if text and _FULL_SHA.match(text) else None 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 new_session_id() -> str: return f"{os.getpid()}-{uuid.uuid4().hex[:12]}" def format_lease_body( *, repo: str, pr_number: int, issue_number: int | None, reviewer_identity: str, profile: str, session_id: str, worktree: str, phase: str, candidate_head: str | None, target_branch: str, target_branch_sha: str | None, last_activity: datetime | None = None, expires_at: datetime | None = None, blocker: str = "none", ) -> str: now = last_activity or datetime.now(timezone.utc) expires = expires_at or (now + timedelta(minutes=DEFAULT_LEASE_TTL_MINUTES)) last_text = now.astimezone(timezone.utc).replace(microsecond=0).isoformat().replace( "+00:00", "Z" ) expires_text = expires.astimezone(timezone.utc).replace(microsecond=0).isoformat().replace( "+00:00", "Z" ) issue_text = f"#{issue_number}" if issue_number else "none" lines = [ MARKER, f"repo: {repo}", f"pr: #{pr_number}", f"issue: {issue_text}", f"reviewer_identity: {reviewer_identity}", f"profile: {profile}", f"session_id: {session_id}", f"worktree: {worktree}", f"phase: {phase}", f"candidate_head: {candidate_head or 'none'}", f"target_branch: {target_branch}", f"target_branch_sha: {target_branch_sha or 'none'}", f"last_activity: {last_text}", f"expires_at: {expires_text}", f"blocker: {blocker}", ] return "\n".join(lines) def parse_lease_comment(body: str) -> dict[str, Any] | None: text = body or "" if MARKER not in text: return None fields: dict[str, str] = {} for match in _FIELD_RE.finditer(text): fields[match.group(1).strip().lower()] = match.group(2).strip() if not fields: return None return { "repo": fields.get("repo"), "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"), "session_id": fields.get("session_id"), "worktree": fields.get("worktree"), "phase": (fields.get("phase") or "").strip().lower() or None, "candidate_head": _normalize_sha(fields.get("candidate_head")), "target_branch": fields.get("target_branch"), "target_branch_sha": _normalize_sha(fields.get("target_branch_sha")), "last_activity": fields.get("last_activity"), "expires_at": fields.get("expires_at"), "blocker": fields.get("blocker"), "raw_fields": fields, } def _lease_entries(comments: list[dict], *, pr_number: int) -> list[dict]: entries: list[dict] = [] for comment in comments or []: parsed = parse_lease_comment(comment.get("body") or "") if not parsed: continue if parsed.get("pr_number") not in (None, pr_number): continue entries.append({ **parsed, "comment_id": comment.get("id"), "author": (comment.get("user") or {}).get("login") or comment.get("author"), "created_at": comment.get("created_at"), "updated_at": comment.get("updated_at"), }) return entries def _lease_expired(lease: dict, *, now: datetime) -> bool: expires_at = _parse_timestamp(lease.get("expires_at")) return bool(expires_at and expires_at <= now) def _minutes_since_activity(lease: dict, *, now: datetime) -> float | None: last = _parse_timestamp(lease.get("last_activity")) if not last: return None return (now - last).total_seconds() / 60.0 def classify_lease_freshness(lease: dict, *, now: datetime | None = None) -> str: """Return active, stale_warning, reclaimable, expired, or terminal.""" now = now or datetime.now(timezone.utc) phase = (lease.get("phase") or "").strip().lower() if phase in _TERMINAL_PHASES: return "terminal" if _lease_expired(lease, now=now): return "expired" minutes = _minutes_since_activity(lease, now=now) if minutes is None: return "active" if minutes >= RECLAIMABLE_MINUTES: return "reclaimable" if minutes >= STALE_WARNING_MINUTES: return "stale_warning" return "active" def find_active_reviewer_lease( comments: list[dict], *, pr_number: int, now: datetime | None = None, ) -> dict[str, Any] | None: """Newest non-terminal, unexpired lease for *pr_number*.""" now = now or datetime.now(timezone.utc) for lease in reversed(_lease_entries(comments, pr_number=pr_number)): phase = (lease.get("phase") or "").strip().lower() if phase in _TERMINAL_PHASES: continue if _lease_expired(lease, now=now): continue if phase in _ACTIVE_PHASES or phase: lease = dict(lease) lease["freshness"] = classify_lease_freshness(lease, now=now) return lease return None def assess_acquire_lease( comments: list[dict], *, pr_number: int, reviewer_identity: str, profile: str, session_id: str, repo: str, issue_number: int | None, worktree: str, candidate_head: str | None, target_branch: str, target_branch_sha: str | None, now: datetime | None = None, ) -> dict[str, Any]: """Fail closed when another session holds an active lease.""" now = now or datetime.now(timezone.utc) reasons: list[str] = [] existing = find_active_reviewer_lease(comments, pr_number=pr_number, now=now) if existing: owner_session = (existing.get("session_id") or "").strip() freshness = existing.get("freshness") or classify_lease_freshness(existing, now=now) if owner_session and owner_session != session_id and freshness in { "active", "stale_warning" }: reasons.append( f"PR #{pr_number} already has active reviewer lease " f"(session_id={owner_session}, phase={existing.get('phase')})" ) elif owner_session and owner_session != session_id and freshness == "reclaimable": reasons.append( f"PR #{pr_number} lease is reclaimable but still held by " f"session_id={owner_session}; explicit reclaim not implemented " "(fail closed)" ) if not (reviewer_identity or "").strip(): reasons.append("reviewer identity required for lease acquisition") if not (session_id or "").strip(): reasons.append("session_id required for lease acquisition") if not (worktree or "").strip(): reasons.append("worktree path required for lease acquisition") allowed = not reasons body = None if allowed: body = format_lease_body( repo=repo, pr_number=pr_number, issue_number=issue_number, reviewer_identity=reviewer_identity, profile=profile, session_id=session_id, worktree=worktree, phase="claimed", candidate_head=candidate_head, target_branch=target_branch, target_branch_sha=target_branch_sha, last_activity=now, ) return { "acquire_allowed": allowed, "reasons": reasons, "existing_lease": existing, "lease_body": body, "session_id": session_id, } def record_session_lease(lease: dict[str, Any]) -> dict[str, Any]: global _SESSION_LEASE _SESSION_LEASE = dict(lease) return dict(_SESSION_LEASE) def clear_session_lease() -> None: global _SESSION_LEASE _SESSION_LEASE = None def get_session_lease() -> dict[str, Any] | None: return dict(_SESSION_LEASE) if _SESSION_LEASE else None def assess_mutation_lease_gate( *, pr_number: int, comments: list[dict], reviewer_identity: str, session_id: str | None, mutation: str, live_head_sha: str | None, pinned_head_sha: str | None, now: datetime | None = None, ) -> dict[str, Any]: """Reviewer mutations require an owned, current PR lease.""" now = now or datetime.now(timezone.utc) reasons: list[str] = [] session = get_session_lease() active = find_active_reviewer_lease(comments, pr_number=pr_number, now=now) if not session: reasons.append( f"no in-session reviewer lease recorded; acquire via " f"gitea_acquire_reviewer_pr_lease before {mutation}" ) elif session.get("pr_number") != pr_number: reasons.append( f"session lease is for PR #{session.get('pr_number')}, not #{pr_number}" ) elif (session.get("session_id") or "") != (session_id or session.get("session_id")): reasons.append("session lease session_id mismatch (fail closed)") if active: owner = (active.get("session_id") or "").strip() if owner and session_id and owner != session_id: reasons.append( f"active PR lease owned by session_id={owner}; current session " f"cannot {mutation}" ) pinned = _normalize_sha(pinned_head_sha) live = _normalize_sha(live_head_sha) lease_head = active.get("candidate_head") if pinned and live and pinned != live: reasons.append( "PR head changed during lease; stop and re-validate before " f"reviewer {mutation}" ) if lease_head and live and lease_head != live: reasons.append( "live PR head differs from lease candidate_head; refresh lease " f"before {mutation}" ) freshness = active.get("freshness") or classify_lease_freshness(active, now=now) if freshness in {"expired", "reclaimable"}: reasons.append(f"reviewer lease freshness is '{freshness}' (fail closed)") else: reasons.append(f"no active reviewer lease found on PR #{pr_number}") allowed = not reasons return { "mutation_allowed": allowed, "block": not allowed, "reasons": reasons, "active_lease": active, "session_lease": session, } def assess_lease_inventory( comments_by_pr: dict[int, list[dict]], *, now: datetime | None = None, ) -> dict[str, Any]: """Summarize lease states across PR comment threads.""" now = now or datetime.now(timezone.utc) active: list[dict] = [] stale: list[dict] = [] reclaimable: list[dict] = [] for pr_number, comments in (comments_by_pr or {}).items(): lease = find_active_reviewer_lease(comments, pr_number=pr_number, now=now) if not lease: continue freshness = lease.get("freshness") or classify_lease_freshness(lease, now=now) entry = {"pr_number": pr_number, "session_id": lease.get("session_id"), "freshness": freshness} if freshness == "stale_warning": stale.append(entry) elif freshness == "reclaimable": reclaimable.append(entry) else: active.append(entry) return { "active_review_leases": active, "stale_review_leases": stale, "reclaimable_review_leases": reclaimable, }